使用 Apache Camel 实现消费 RabbitMQ 消息并通过 SMPP 协议发送短消息
Apache Camel 是一个开源集成框架,使您能够快速轻松地集成各种消费或生产数据的系统。[1]
这里使用 Apache Camel 来实现消费 RabbitMQ 的消息,并将其通过 SMPP 协议发送短信息的功能。第一次使用,了解不多,但总体感觉确实集成的很彻底,使用起来很方便。
主要功能的代码如下:
package me.liujiajia.sample.samplesmpp;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.model.dataformat.JsonLibrary;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import static org.apache.camel.component.smpp.SmppConstants.DEST_ADDR;
import static org.apache.camel.component.smpp.SmppConstants.SOURCE_ADDR;
@Component
class Mq2SmsRoute extends RouteBuilder {
@Autowired
private ShortMsgFormatService shortMsgFormatService;
@Autowired
private SMSProcessor smsProcessor;
@Override
public void configure() {
from("rabbitmq:sms" +
"?queue=sms" +
// 也可以在这里配置 RabbitMQ 服务器的地址、用户和密码
// "&addresses=localhost:5672&username=sms&password=123456" +
"&autoDelete=false&autoAck=false" +
// 如果配置了死信队列,发送失败的消息会自动转发到配置的死信队列
"&deadLetterExchange=sms-failed&deadLetterQueue=sms-failed")
.log("mq >>> ${body}")
// 使用 Jackson 反序列化
.unmarshal()
.json(JsonLibrary.Jackson, ShortMsgEntity.class)
.log("jp >>> ${body.id}")
.log("jp >>> ${body.name}")
.process(exchange -> {
ShortMsgEntity shortMsg = exchange.getIn().getBody(ShortMsgEntity.class);
// 调用 service 示例
shortMsgFormatService.format(shortMsg);
// 可以通过 setHeader 方法指定 Header
// exchange.getIn().setHeader(SERVICE_TYPE, "CMT");
// exchange.getIn().setHeader(SOURCE_ADDR_TON, TypeOfNumber.ALPHANUMERIC.value());
// exchange.getIn().setHeader(SOURCE_ADDR_NPI, NumberingPlanIndicator.UNKNOWN.value());
exchange.getIn().setHeader(SOURCE_ADDR, shortMsg.getFrom());
// exchange.getIn().setHeader(DEST_ADDR_TON, TypeOfNumber.ALPHANUMERIC.value());
// exchange.getIn().setHeader(DEST_ADDR_NPI, NumberingPlanIndicator.UNKNOWN.value());
exchange.getIn().setHeader(DEST_ADDR, shortMsg.getTo());
// 设置消息内容
exchange.getIn().setBody(shortMsg.getContent());
})
.log("to >>> ${body}")
// 默认配置下,如果和 SMPP 服务器直接的链接断了,会自动尝试重连,直至成功为止(默认的最大重连次数很大)。
// 断连过程中接收的消息会发送失败。
.to("smpp://sms@localhost:2775" +
"?password=123456" +
"&enquireLinkTimer=3000" +
"&transactionTimer=5000" +
"&systemType=producer"
)
// processor 示例
// 运行到这里的时候,已经可以从 Header 中获取 msgid(在 messages.log 文件中为 queue-msgid)
// exchange.getIn().getHeader(SmppConstants.ID, String.class)
// 如果发送处理中发生异常(比如连接不上 SMPP 服务器),代码不进行到这里的 processor
.process(smsProcessor)
// 这里的 body 仍然和 to 之前一样,没有变化
.log("ed >>> ${body}");
}
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
从上面的代码可以看到,除了 from()
和 to()
之外,最重要的就是 process()
方法,它接收一个 Processor
参数。
package org.apache.camel;
@FunctionalInterface
public interface Processor {
void process(Exchange exchange) throws Exception;
}
2
3
4
5
6
Processor
仅包含一个 process()
方法,方法中可以通过 exchange.getIn()
获取当前的消息,之后就可以对其进行设置了。
本示例中使用了 3 个组件:RabbitMQ [2]、JSON Jackson [3] 和 SMPP [4] 。
每个组件的文档基本都包含如下几个部分:
URI FORMAT | URI 格式
这部分是描述资源路径的格式,和网页地址比较类似。
如下是 RabbitMQ 组件的 URI 格式。javascriptrabbitmq:exchangeName?[options]
1其中
[options]
的具体参数见下面的 ENDPOINT OPTIONS 部分。COMPONENT OPTIONS | 组件选项
组件的选项一般和下面的端点选项(ENDPOINT OPTIONS)一般都是一样的,如果两个都配置的话,端点选项的优先级较高。
组件选项示例:
yamlcamel: component: rabbitmq: addresses: localhost:5672 username: sms password: 123456 smpp: enquire-link-timer: 3000
1
2
3
4
5
6
7
8ENDPOINT OPTIONS | 端点选项
用来指定单个端点的选项,拼在 URI 中
?
的后面即可。MESSAGE HEADERS | 消息头
当前组件消息支持的 Header,可以通过
exchange.getIn().setHeader()
方法进行设置。SAMPLES | 示例
这里是组件的写法示例。
虽然是第一次用,但是组件的文档描述的都很详细,选项也比较容易理解。不过组件的选项大都比较多,使用前最好还是花点时间仔细看一下。
关于组件的版本,虽然现在最新的是 4.1.0 ,但由于部分组件还没有更新到这个版本,为避免版本不一致可能导致的问题,这里选用了 3.21.2 版。
另外官方文档建议使用 Java 的 LTS 版本(11 或 17)[5]。
pom.xml 文件内容如下。
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.7.17</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>me.liujiajia.sample</groupId>
<artifactId>sample-smpp</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>sample-smpp</name>
<description>Demo project for Spring Boot</description>
<properties>
<java.version>17</java.version>
<!--<camel.version>4.1.0</camel.version>-->
<camel.version>3.21.2</camel.version>
</properties>
<dependencies>
<!-- https://mvnrepository.com/artifact/org.apache.camel.springboot/camel-spring-boot-starter -->
<dependency>
<groupId>org.apache.camel.springboot</groupId>
<artifactId>camel-spring-boot-starter</artifactId>
<version>${camel.version}</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.camel.springboot/camel-smpp-starter -->
<dependency>
<groupId>org.apache.camel.springboot</groupId>
<artifactId>camel-smpp-starter</artifactId>
<version>${camel.version}</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.camel.springboot/camel-jackson-starter -->
<dependency>
<groupId>org.apache.camel.springboot</groupId>
<artifactId>camel-jackson-starter</artifactId>
<version>${camel.version}</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.camel.springboot/camel-rabbitmq-starter -->
<dependency>
<groupId>org.apache.camel.springboot</groupId>
<artifactId>camel-rabbitmq-starter</artifactId>
<version>${camel.version}</version>
</dependency>
<!-- Test -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
附 1. 相关代码
Mq2SmsRoute
中使用的几个自定义的类都比较简单,仅作为示例展示。为方便理解,也一并贴一下。
ShortMsgFormatService
package me.liujiajia.sample.samplesmpp;
public interface ShortMsgFormatService {
void format(ShortMsgEntity bodyIn);
}
2
3
4
5
ShortMsgFormatServiceImpl
package me.liujiajia.sample.samplesmpp;
import org.springframework.stereotype.Service;
@Service
public class ShortMsgFormatServiceImpl implements ShortMsgFormatService {
@Override
public void format(ShortMsgEntity msg) {
msg.setContent("%s, %s.".formatted(msg.getContent(), msg.getName()));
}
}
2
3
4
5
6
7
8
9
10
11
12
SMSProcessor
package me.liujiajia.sample.samplesmpp;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.component.smpp.SmppConstants;
import org.springframework.stereotype.Component;
@Component
public class SMSProcessor implements Processor {
@Override
public void process(Exchange exchange) throws Exception {
Message m = exchange.getIn();
System.out.println(m.getBody());
System.out.println(m.getHeader(SmppConstants.ID, String.class));
System.out.println(m.getHeaders());
}
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
ShortMsgEntity
package me.liujiajia.sample.samplesmpp;
public class ShortMsgEntity {
private Integer id;
private String name;
private String from;
private String to;
private String content;
public Integer getId() {
return id;
}
public void setId(Integer id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String getFrom() {
return from;
}
public void setFrom(String from) {
this.from = from;
}
public String getTo() {
return to;
}
public void setTo(String to) {
this.to = to;
}
public String getContent() {
return content;
}
public void setContent(String content) {
this.content = content;
}
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
application.yml
camel:
springboot:
name: SMS Service
component:
rabbitmq:
addresses: localhost:5672
username: sms
password: 123456
# smpp:
# enquire-link-timer: 3000
2
3
4
5
6
7
8
9
10
附 2. 本地开发用的 docker-compose 文件
RabbitMQ
version: '3.0'
services:
rabbitmq:
image: rabbitmq:3-management
environment:
- RABBITMQ_DEFAULT_USER=guest
- RABBITMQ_DEFAULT_PASS=guest
- RABBITMQ_VM_MEMORY_HIGH_WATERMARK_RELATIVE=0.8
ports:
- "5672:5672"
- "15672:15672"
volumes:
- ./src/rabbitmq/20-mem.conf:/etc/rabbitmq/conf.d/20-mem.conf
2
3
4
5
6
7
8
9
10
11
12
13
SMPP Server
为了在本地尝试搭建 SMPP 服务器,找了好几种方式,最终在本地能运行起来的只有 Jasmin [6] 。
version: "3.10"
services:
redis:
image: redis:alpine
restart: unless-stopped
healthcheck:
test: redis-cli ping | grep PONG
deploy:
resources:
limits:
cpus: '0.2'
memory: 128M
security_opt:
- no-new-privileges:true
rabbit-mq:
image: rabbitmq:3.10-management-alpine
restart: unless-stopped
healthcheck:
test: rabbitmq-diagnostics -q ping
deploy:
resources:
limits:
cpus: '0.5'
memory: 525M
security_opt:
- no-new-privileges:true
jasmin:
image: jookies/jasmin:latest
restart: unless-stopped
ports:
- 2775:2775
- 8990:8990
- 1401:1401
depends_on:
redis:
condition: service_healthy
rabbit-mq:
condition: service_healthy
environment:
REDIS_CLIENT_HOST: redis
AMQP_BROKER_HOST: rabbit-mq
deploy:
resources:
limits:
cpus: '1'
memory: 256M
security_opt:
- no-new-privileges:true
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
启动后配置用户
需要通过 telnet
连接 jcli
工具来创建用户 [7] 。
telnet 127.0.0.1 8990
Windows 用户的需要启用 Telnet 客户端功能:
- 按
WIN + R
快捷键,输入appwiz.cpl
打开 程序和功能 页面; - 点击左侧的 启用或关闭 Windows 功能,勾选 Telnet 客户端,然后确定。
Windows 下的 Telnet 虽然能用,但是每次回车后都需要再任一输入一个字符才能看到响应,不知道是不是只有这个
jcli
才会这样,使用起来很不方便。
总的命令汇总如下,全部复制然后直接粘贴就可以了。
smppccm -a
cid DEMO_CONNECTOR
host 127.0.0.1
port 2775
username sms
password 123456
submit_throughput 110
ok
smppccm -1 DEMO_CONNECTOR
smppccm --list
mtrouter -a
type defaultroute
connector smppc(DEMO_CONNECTOR)
rate 0.00
ok
group -a
gid smsgroup
ok
user -a
username sms
password 123456
gid smsgroup
uid sms
ok
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
注意:
Jasmin 的 Pod 每次重启后用户信息都会丢失,需要重新创建。
发送消息
创建用户后可以通过点击如下链接发送测试消息:
http://127.0.0.1:1401/send?username=sms&password=123456&to=06222172&content=hello
能看到类似 Success "0ae1613a-7fbf-409f-b3d2-efd24fc49ad7" 的响应则说明用户创建成功了。
通过 HTTP 发送的消息会记录在日志文件 /var/log/jasmin/http-api.log
中,如果是通过代码发送的消息,则可以查看 /var/log/jasmin/messages.log
日志文件。
可以在进入 Pod 后执行如下命令实时查看 messages.log 的内容。
tail -fn100 /var/log/jasmin/messages.log
https://camel.apache.org/components/3.21.x/rabbitmq-component.html ↩︎
https://camel.apache.org/components/3.21.x/dataformats/jackson-dataformat.html ↩︎
https://camel.apache.org/components/3.21.x/smpp-component.html ↩︎
https://camel.apache.org/camel-core/getting-started/index.html#BookGettingStarted-CreatingYourFirstProject ↩︎
https://docs.jasminsms.com/en/latest/installation/index.html#docker ↩︎
https://docs.jasminsms.com/en/latest/installation/index.html#sending-your-first-sms ↩︎