使用 Apache Camel 实现消费 RabbitMQ 消息并通过 SMPP 协议发送短消息

🏷️ Camel 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}");
    }
}

从上面的代码可以看到，除了 from() 和 to() 之外，最重要的就是 process() 方法，它接收一个 Processor 参数。

package org.apache.camel;

@FunctionalInterface
public interface Processor {
    void process(Exchange exchange) throws Exception;
}

Processor 仅包含一个 process() 方法，方法中可以通过 exchange.getIn() 获取当前的消息，之后就可以对其进行设置了。

本示例中使用了 3 个组件：RabbitMQ ^[2]、JSON Jackson ^[3] 和 SMPP ^[4] 。

每个组件的文档基本都包含如下几个部分：

• URI FORMAT | URI 格式

  这部分是描述资源路径的格式，和网页地址比较类似。
  如下是 RabbitMQ 组件的 URI 格式。

  rabbitmq:exchangeName?[options]

  其中 [options] 的具体参数见下面的 ENDPOINT OPTIONS 部分。

• COMPONENT OPTIONS | 组件选项

  组件的选项一般和下面的端点选项（ENDPOINT OPTIONS）一般都是一样的，如果两个都配置的话，端点选项的优先级较高。

  组件选项示例：

  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）^[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>

附 1. 相关代码

Mq2SmsRoute 中使用的几个自定义的类都比较简单，仅作为示例展示。为方便理解，也一并贴一下。

ShortMsgFormatService

package me.liujiajia.sample.samplesmpp;

public interface ShortMsgFormatService {
    void format(ShortMsgEntity bodyIn);
}

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()));
    }
}

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());
    }
}

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;
    }
}

application.yml

camel:
  springboot:
    name: SMS Service
  component:
    rabbitmq:
      addresses: localhost:5672
      username: sms
      password: 123456
#    smpp:
#      enquire-link-timer: 3000

附 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

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

启动后配置用户

需要通过 telnet 连接 jcli 工具来创建用户 ^[7] 。

telnet 127.0.0.1 8990

Windows 用户的需要启用 Telnet 客户端功能：

1. 按 WIN + R 快捷键，输入 appwiz.cpl 打开 程序和功能 页面；
2. 点击左侧的 启用或关闭 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

注意：
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

---

1. https://camel.apache.org/ ↩︎

2. https://camel.apache.org/components/3.21.x/rabbitmq-component.html ↩︎

3. https://camel.apache.org/components/3.21.x/dataformats/jackson-dataformat.html ↩︎

4. https://camel.apache.org/components/3.21.x/smpp-component.html ↩︎

5. https://camel.apache.org/camel-core/getting-started/index.html#BookGettingStarted-CreatingYourFirstProject ↩︎

6. https://docs.jasminsms.com/en/latest/installation/index.html#docker ↩︎

7. https://docs.jasminsms.com/en/latest/installation/index.html#sending-your-first-sms ↩︎