Remoting

RocketMQ在5.0后为了适应多个语言以及考虑到性能等问题，引入了GRPC，不过5.0中依然保留了对Remoting协议的支持，5.0的新组件Proxy中包含了支持Remoting和GRPC两个端口，使用Remoting作为通讯协议的话，可以使用已经支持Spring Boot的客户端版本

依赖

1
2
3
4
5
6
7
8
<!-- Apache RocketMQ存在远程命令执行高风险漏洞，漏洞编号为CVE-2023-33246     -->
<!-- 影响版本: 5.0.0 <= Apache RocketMQ < 5.1.1 4.0.0 <= Apache RocketMQ < 4.9.6  -->
<!-- 和RabbitMQ、Kafka不同，Spring没有集成RocketMQ，rocketmq-spring-boot-starter是RocketMQ集成了Spring Boot以适应Spring Boot环境，集成了Spring Messaging  -->
<dependency>
  <groupId>org.apache.rocketmq</groupId>
  <artifactId>rocketmq-spring-boot-starter</artifactId>
  <version>2.3.0</version>
</dependency>

生产者

配置

1
2
3
4
rocketmq:
  name-server: 127.0.0.1:9876
  producer:
    group: test-group

上面是两个生产者必须的配置，缺一不可，这是因为RocketMQTemplate实例化必需要这两个参数，具体可以参考org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration

代码

1
2
3
4
5
6
7
8
9
10
11
public class MessageService {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void sendMessage(String message){
        Message<String> messaging = MessageBuilder.withPayload(message).build();
        rocketMQTemplate.send(RocketConfig.TEST_TOPIC, messaging);
    }

}

一般情况下，直接使用rocketMQTemplate发送消息即可，支持单方向发送，同步发送，异步发送，这里的Message是spring-messaging的内容，spring-rabbitmq和spring-kafka都使用到了这个类

消费者

配置

1
@RocketMQMessageListener(consumerGroup = "test-group", topic = RocketConfig.TEST_TOPIC)

@RocketMQMessageListener注解中可以配置单个监听器的部分配置

1
2
3
4
5
rocketmq:
  name-server: 127.0.0.1:9876
  consumer:
    pull-batch-size: 10
    topic: test-group

yaml中配置消费者的全局配置，并且其中大部分配置只支持通过yaml的方式配置

代码

1
2
3
4
5
6
7
8
9
10
@Component
@RocketMQMessageListener(consumerGroup = "test-group", topic = RocketConfig.TEST_TOPIC)
public class MessageListener implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        System.out.println(message);
    }

}

消费者直接实现RocketMQListener即可，底层具体逻辑是先将消息转为String，然后再通过org.springframework.messaging.converter.MessageConverter#fromMessage方法转为具体的类型

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
private Object doConvertMessage(MessageExt messageExt) {
    if (!Objects.equals(this.messageType, MessageExt.class) && !Objects.equals(this.messageType, org.apache.rocketmq.common.message.Message.class)) {
        String str = new String(messageExt.getBody(), Charset.forName(this.charset));
        if (Objects.equals(this.messageType, String.class)) {
            return str;
        } else {
            try {
                return this.messageType instanceof Class ? this.getMessageConverter().fromMessage(MessageBuilder.withPayload(str).build(), (Class)this.messageType) : ((SmartMessageConverter)this.getMessageConverter()).fromMessage(MessageBuilder.withPayload(str).build(), (Class)((ParameterizedType)this.messageType).getRawType(), this.methodParameter);
            } catch (Exception var4) {
                log.info("convert failed. str:{}, msgType:{}", str, this.messageType);
                throw new RuntimeException("cannot convert message to " + this.messageType, var4);
            }
        }
    } else {
        return messageExt;
    }
}

事务消息

介绍

官方文档：https://rocketmq.apache.org/zh/docs/featureBehavior/04transactionmessage

RocketMQ的事务消息可以理解为一种半消息，需要通过二次确认才能确定是否可以发送给消费者，分为以下几步

  1. 发送事务消息到RocketMQ
  2. 执行本地事务
  3. 根据本地事务执行情况，提交二次确认到RocketMQ，如果RocketMQ没有收到二次确认，它会主动请求生产者，检查确认，检查确认可以保证二次确认可以成功到达RocketMQ

RocketMQ事务消息保证了发送消息和本地事务执行两个动作状态的一致性，事务消息不支持异步发送

RocketMQ和RabbitMQ的事务不太一样，RocketMQ的事务是针对单条消息，而RabbitMQ和数据库事务一样，针对的是多个操作

代码

RocketMQ的事务消息，就是加上事务监听器，先发送消息，再通过事务监听器发送二次确认，并且事务监听器支持检查确认

1
2
3
4
5
6
7
8
9
10
11
12
@Service
public class MessageService {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    public void sendMessage(String message){
        Message<String> messaging = MessageBuilder.withPayload(message).build();
        rocketMQTemplate.sendMessageInTransaction(RocketConfig.TEST_TOPIC, messaging, null);
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
@RocketMQTransactionListener
public class TransactionListener implements RocketMQLocalTransactionListener {

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(org.springframework.messaging.Message message, Object o) {
        System.out.println("运行本地事务");
        return RocketMQLocalTransactionState.COMMIT;
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(org.springframework.messaging.Message message) {
        return RocketMQLocalTransactionState.COMMIT;
    }
}

加上@RocketMQTransactionListener注解，就是全局定义事务监听器
默认会给所有名称为“rocketMQTemplate”的RocketMQTemplate的bean实例设置这个事务监听器
并且同一个RocketMQTemplate实例不能设置两个RocketMQLocalTransactionListener，不然会报错

由于这里RocketMQTemplate和DefaultMQProducer都交给Spring管理了，设置多个事务监听器不太方便
如果需要为不同的业务设置不同的RocketMQ事务监听器，个人认为定义一个事务监听器，然后在这个事务监听器中对不同的业务进行不同的处理
或者通过rocketMQTemplate.getProducer().getDefaultMQProducerImpl().sendMessageInTransaction这种方式发送消息并指定事务监听器，其中的getDefaultMQProducerImpl已经被标注为已弃用了，不过还可以将rocketMQTemplate.getProducer()作为参数，调用new DefaultMQProducerImpl(producer)的构造方法构造一个新的DefaultMQProducerImpl实例

延申

为了保证一致性，还有一个思路，不使用事务消息，不过将发送消息这一步放到业务的最后一步，发送成功即提交本地事务，发送失败即回滚本地事务，但是有几个问题

  1. 如果本地事务执行时间非常久，相应的回滚也会非常久，如果发送消息失败而导致业务回滚，那么对性能是极大的浪费，而使用事务消息没有执行顺序的限制，完全可以将发送消息靠前执行
  2. 如果发送消息时间非常久，那么本地事务和发送消息两个步骤顺序执行会很费时间，而使用事务消息，两个步骤异步执行也可以保证最后的一致性

GRPC

依赖

对于GRPC，官方client并未提供Spring Bean方式接入的示例，只能使用原生的client，在使用过程中，可以直接使用或者自己嵌入到Spring

1
2
3
4
5
<dependency>
  <groupId>org.apache.rocketmq</groupId>
  <artifactId>rocketmq-client-java</artifactId>
  <version>5.0.7</version>
</dependency>

生产者

由于是原生的client，所以没有Spring相关配置

代码

org.apache.rocketmq.client.java.example包中提供了一些构建生产者，发送消息的示例，下面是其中一个构造生产者和发送消息的一个例子，这里需要注意，使用GRPC的话，必须配置Proxy的GRPC端口，不再是NameServer的地址了

1
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
private static Producer buildProducer(TransactionChecker checker, String... topics) throws ClientException {
    ClientServiceProvider provider = ClientServiceProvider.loadService();
    SessionCredentialsProvider sessionCredentialsProvider = new StaticSessionCredentialsProvider("yourAccessKey", "yourSecretKey");
    ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder().setEndpoints("foobar.com:8080").setCredentialProvider(sessionCredentialsProvider).build();
    ProducerBuilder builder = provider.newProducerBuilder().setClientConfiguration(clientConfiguration).setTopics(topics);
    if (checker != null) {
        builder.setTransactionChecker(checker);
    }

    return builder.build();
}

public static void main(String[] args) throws ClientException {
    ClientServiceProvider provider = ClientServiceProvider.loadService();
    String topic = "yourNormalTopic";
    Producer producer = ProducerSingleton.getInstance(new String[]{topic});
    byte[] body = "This is a normal message for Apache RocketMQ".getBytes(StandardCharsets.UTF_8);
    String tag = "yourMessageTagA";
    Message message = provider.newMessageBuilder().setTopic(topic).setTag(tag).setKeys(new String[]{"yourMessageKey-1c151062f96e"}).setBody(body).build();

    try {
        SendReceipt sendReceipt = producer.send(message);
        log.info("Send message successfully, messageId={}", sendReceipt.getMessageId());
    } catch (Throwable var8) {
        log.error("Failed to send message", var8);
    }

}

消费者

由于是原生的client，所以没有Spring相关配置

代码

和生产者一样，org.apache.rocketmq.client.java.example包总也有很多消费者的示例，包括推模式，拉模式，同消费，异步消费等，下面是一个推模式同步消费的一个例子，使用GRPC的话，必须配置Proxy的GRPC端口，不再是NameServer的地址了

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
public static void main(String[] args) throws ClientException, InterruptedException, IOException {
    ClientServiceProvider provider = ClientServiceProvider.loadService();
    String accessKey = "yourAccessKey";
    String secretKey = "yourSecretKey";
    SessionCredentialsProvider sessionCredentialsProvider = new StaticSessionCredentialsProvider(accessKey, secretKey);
    String endpoints = "foobar.com:8080";
    ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder().setEndpoints(endpoints).setCredentialProvider(sessionCredentialsProvider).build();
    String tag = "yourMessageTagA";
    FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
    String consumerGroup = "yourConsumerGroup";
    String topic = "yourTopic";
    PushConsumer pushConsumer = provider.newPushConsumerBuilder().setClientConfiguration(clientConfiguration).setConsumerGroup(consumerGroup).setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression)).setMessageListener((messageView) -> {
        log.info("Consume message={}", messageView);
        return ConsumeResult.SUCCESS;
    }).build();
    Thread.sleep(Long.MAX_VALUE);
    pushConsumer.close();
}

事务消息

代码

下面是org.apache.rocketmq.client.java.example包中对于事务消息的一个例子

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
public static void main(String[] args) throws ClientException {
    ClientServiceProvider provider = ClientServiceProvider.loadService();
    String topic = "yourTransactionTopic";
    TransactionChecker checker = (messageView) -> {
        log.info("Receive transactional message check, message={}", messageView);
        return TransactionResolution.COMMIT;
    };
    Producer producer = ProducerSingleton.getTransactionalInstance(checker, new String[]{topic});
    Transaction transaction = producer.beginTransaction();
    byte[] body = "This is a transaction message for Apache RocketMQ".getBytes(StandardCharsets.UTF_8);
    String tag = "yourMessageTagA";
    Message message = provider.newMessageBuilder().setTopic(topic).setTag(tag).setKeys(new String[]{"yourMessageKey-565ef26f5727"}).setBody(body).build();

    try {
        SendReceipt sendReceipt = producer.send(message, transaction);
        log.info("Send transaction message successfully, messageId={}", sendReceipt.getMessageId());
    } catch (Throwable var10) {
        log.error("Failed to send message", var10);
        return;
    }

    transaction.commit();
}

这段代码可以看出来，这里的事务消息实现了多消息事务，它的实现原理是每次发送消息后，将消息保存下来，最后commit的时候，将每条事务消息依次提交，通过这种方式实现了类似于RabbitMQ以及数据库的事务机制，实现了多条消息同时提交或者同时回滚，下面是commit方法的代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public void commit() throws ClientException {
    if (this.messageSendReceiptMap.isEmpty()) {
        throw new IllegalStateException("Transactional message has not been sent yet");
    } else {
        Iterator var1 = this.messageSendReceiptMap.entrySet().iterator();

        while(var1.hasNext()) {
            Map.Entry<PublishingMessageImpl, SendReceiptImpl> entry = (Map.Entry)var1.next();
            PublishingMessageImpl publishingMessage = (PublishingMessageImpl)entry.getKey();
            SendReceiptImpl sendReceipt = (SendReceiptImpl)entry.getValue();
            this.producerImpl.endTransaction(sendReceipt.getEndpoints(), new GeneralMessageImpl(publishingMessage), sendReceipt.getMessageId(), sendReceipt.getTransactionId(), TransactionResolution.COMMIT, TransactionSource.SOURCE_CLIENT);
        }

    }
}

测试

这里主要是看一下是否用到了GRPC

可以看出来，确实使用到了GRPC

本站总访问量：次本站总访问数：人
本站由 Peng Tong 使用 Stellar 1.29.1 主题创建。
本博客所有文章除特别声明外，均采用 CC BY-NC-SA 4.0 许可协议，转载请注明出处。