Apache Camel 是一个开源集成框架，使您能够快速轻松地集成各种消费或生产数据的系统。[^camel] 这里使用 Apache Camel 来实现消费 RabbitMQ 的消息，并将其通过 SMPP 协议发送短信息的功能。第一次使用，了解不多，但总体感觉确实集成的很彻底，使用起来很方便。主要功能的代码如下： ```java 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}"); } } ``` 从上面的代码可以看到，除了 `from()` 和 `to()` 之外，最重要的就是 `process()` 方法，它接收一个 `Processor` 参数。 ```java package org.apache.camel; @FunctionalInterface public interface Processor { void process(Exchange exchange) throws Exception; } ``` `Processor` 仅包含一个 `process()` 方法，方法中可以通过 `exchange.getIn()` 获取当前的消息，之后就可以对其进行设置了。本示例中使用了 3 个组件：RabbitMQ [^camel-rabbitmq]、JSON Jackson [^camel-jackson] 和 SMPP [^camel-smpp] 。每个组件的文档基本都包含如下几个部分： - URI FORMAT | URI 格式这部分是描述资源路径的格式，和网页地址比较类似。如下是 RabbitMQ 组件的 URI 格式。 ```javascript rabbitmq:exchangeName?[options] ``` 其中 `[options]` 的具体参数见下面的 ENDPOINT OPTIONS 部分。 - COMPONENT OPTIONS | 组件选项组件的选项一般和下面的端点选项（ENDPOINT OPTIONS）一般都是一样的，如果两个都配置的话，端点选项的优先级较高。组件选项示例： ```yaml camel: component: rabbitmq: addresses: localhost:5672 username: sms password: 123456 smpp: enquire-link-timer: 3000 ``` - ENDPOINT OPTIONS | 端点选项用来指定单个端点的选项，拼在 URI 中 `?` 的后面即可。 - MESSAGE HEADERS | 消息头当前组件消息支持的 Header，可以通过 `exchange.getIn().setHeader()` 方法进行设置。 - SAMPLES | 示例这里是组件的写法示例。虽然是第一次用，但是组件的文档描述的都很详细，选项也比较容易理解。不过组件的选项大都比较多，使用前最好还是花点时间仔细看一下。关于组件的版本，虽然现在最新的是 **4.1.0** ，但由于部分组件还没有更新到这个版本，为避免版本不一致可能导致的问题，这里选用了 **3.21.2** 版。另外官方文档建议使用 Java 的 LTS 版本（11 或 17）[^firstproject]。 pom.xml 文件内容如下。 ```xml 4.0.0 org.springframework.boot spring-boot-starter-parent 2.7.17 me.liujiajia.sample sample-smpp 0.0.1-SNAPSHOT sample-smpp Demo project for Spring Boot 17 3.21.2 org.apache.camel.springboot camel-spring-boot-starter ${camel.version} org.apache.camel.springboot camel-smpp-starter ${camel.version} org.apache.camel.springboot camel-jackson-starter ${camel.version} org.apache.camel.springboot camel-rabbitmq-starter ${camel.version} org.springframework.boot spring-boot-starter-test test org.springframework.boot spring-boot-maven-plugin ``` ## 附 1. 相关代码 `Mq2SmsRoute` 中使用的几个自定义的类都比较简单，仅作为示例展示。为方便理解，也一并贴一下。 **ShortMsgFormatService** ```java package me.liujiajia.sample.samplesmpp; public interface ShortMsgFormatService { void format(ShortMsgEntity bodyIn); } ``` **ShortMsgFormatServiceImpl** ```java 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())); } } ``` **SMSProcessor** ```java 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()); } } ``` **ShortMsgEntity** ```java 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; } } ``` **application.yml** ```yaml camel: springboot: name: SMS Service component: rabbitmq: addresses: localhost:5672 username: sms password: 123456 # smpp: # enquire-link-timer: 3000 ``` ## 附 2. 本地开发用的 docker-compose 文件 ### RabbitMQ ```yaml 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 ``` ### SMPP Server 为了在本地尝试搭建 SMPP 服务器，找了好几种方式，最终在本地能运行起来的只有 Jasmin [^docker-installtion-jasmin] 。 ```yaml 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 ``` #### 启动后配置用户需要通过 `telnet` 连接 `jcli` 工具来创建用户 [^sending-your-first-sms] 。 ```bash telnet 127.0.0.1 8990 ``` Windows 用户的需要启用 Telnet 客户端功能： 1. 按 `WIN + R` 快捷键，输入 `appwiz.cpl` 打开程序和功能页面； 2. 点击左侧的启用或关闭 Windows 功能，勾选 Telnet 客户端，然后确定。 > Windows 下的 Telnet 虽然能用，但是每次回车后都需要再任一输入一个字符才能看到响应，不知道是不是只有这个 `jcli` 才会这样，使用起来很不方便。总的命令汇总如下，全部复制然后直接粘贴就可以了。 ```bash 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 ``` > **注意：** > Jasmin 的 Pod 每次重启后用户信息都会丢失，需要重新创建。 #### 发送消息创建用户后可以通过点击如下链接发送测试消息： [http://127.0.0.1:1401/send?username=sms&password=123456&to=06222172&content=hello](http://127.0.0.1:1401/send?username=sms&password=123456&to=06222172&content=hello "点击发送消息到 Jasmin") 能看到类似 *Success "0ae1613a-7fbf-409f-b3d2-efd24fc49ad7"* 的响应则说明用户创建成功了。通过 HTTP 发送的消息会记录在日志文件 `/var/log/jasmin/http-api.log` 中，如果是通过代码发送的消息，则可以查看 `/var/log/jasmin/messages.log` 日志文件。可以在进入 Pod 后执行如下命令实时查看 *messages.log* 的内容。 ```bash tail -fn100 /var/log/jasmin/messages.log ``` [^camel]: [https://camel.apache.org/](https://camel.apache.org/ "Apache Camel") [^camel-rabbitmq]: [https://camel.apache.org/components/3.21.x/rabbitmq-component.html](https://camel.apache.org/components/3.21.x/rabbitmq-component.html "RabbitMQ :: Apache Camel") [^camel-jackson]: [https://camel.apache.org/components/3.21.x/dataformats/jackson-dataformat.html](https://camel.apache.org/components/3.21.x/dataformats/jackson-dataformat.html "JSON Jackson :: Apache Camel") [^camel-smpp]: [https://camel.apache.org/components/3.21.x/smpp-component.html](https://camel.apache.org/components/3.21.x/smpp-component.html "SMPP :: Apache Camel") [^firstproject]: [https://camel.apache.org/camel-core/getting-started/index.html#BookGettingStarted-CreatingYourFirstProject](https://camel.apache.org/camel-core/getting-started/index.html#BookGettingStarted-CreatingYourFirstProject "CREATING YOUR FIRST PROJECT") [^docker-installtion-jasmin]: [https://docs.jasminsms.com/en/latest/installation/index.html#docker](https://docs.jasminsms.com/en/latest/installation/index.html#docker "Docker - Installation - Jasmin SMS Gateway") [^sending-your-first-sms]: [https://docs.jasminsms.com/en/latest/installation/index.html#sending-your-first-sms](https://docs.jasminsms.com/en/latest/installation/index.html#sending-your-first-sms "Sending your first SMS - Installation - Jasmin SMS Gateway")

目录