JarCastingrocketmq-wrapper
简介
Rocketmq-wrapper是对rocketmq client library的二次封装，支持普通消息和事务消息的发送和处理。Rocketmq-wrapper能大大方便我们使用rocketmq client来来构建应用程序，而忽略一些细节上的事情。
- 支持同步消息发送
- 支持异步消息发送
- 支持事务消息发送
- 支持顺序消息发送
- 支持rocketmq常用配置
- 支持自定义消息序列化方式
- 支持动态设置监听topic
- 支持多生产者
- 支持多消费者
使用
引入library：
<dependency>
<groupId>io.github.zxgangandy</groupId>
<artifactId>rocketmq-wrapper-core</artifactId>
<version>1.1.2</version>
</dependency>
消息生产者例子：
private RMProducer producer;
@Before
public void init() {
producer = RMWrapper.with(RMProducer.class)
.producerGroup("producer-test")
.nameSrvAddr("127.0.0.1:9876")
.topic("test1").retryTimes(3)
.txListener(new TxListener())
.start();
}
//同步消息
@Test
public void sendMsgSync() {
try {
SendResult sendResult = producer.sendSync("test", new MessageBody().setContent("a"));
System.out.println("sendMsgSync, sendResult=" +sendResult);
} catch (Exception e) {
e.printStackTrace();
}
}
//异步消息
@Test
public void sendMsgAsync() {
try {
producer.sendAsync("test", new MessageBody().setContent("b"), new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
System.out.println("sendMsgAsync, sendResult=" +sendResult);
}
@Override
public void onException(Throwable e) {
System.out.println("sendMsgAsync, e=" +e);
}
});
} catch (Exception e) {
e.printStackTrace();
}
}
//事务消息
@Test
public void sendTxMsg() {
try {
SendResult sendResult = producer.sendTransactional("test", new MessageBody().setContent("c"), "d");
System.out.println("sendTxMsg, sendResult=" +sendResult);
} catch (Exception e) {
e.printStackTrace();
}
}
//顺序消息
@Test
public void sendOrderlyMsg() {
try {
for(int i=0; i<100; i++) {
SendResult sendResult = producer.sendOrderly("test", new MessageBody().setContent("c"), "d");
System.out.println("sendTxMsg, sendResult=" + sendResult);
}
} catch (Exception e) {
e.printStackTrace();
}
}
- 事务消息需要实现TransactionListener接口，在使用rocketmq-wrapper的时候只需要继承AbstractTransactionListener即可；
消费者例子
@Test
public void concurrentlyProcessor() throws InterruptedException {
RMWrapper.with(RMConsumer.class)
.consumerGroup("consumer-test")
.nameSrvAddr("127.0.0.1:9876")
.subscribe("test1")
.concurrentlyProcessor((messageBody) -> {
System.out.println("concurrentlyProcessor, messageBody=" + messageBody);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
})
.start();
Thread.sleep(50000);
}
@Test
public void orderlyRawProcessor() throws InterruptedException {
RMWrapper.with(RMConsumer.class)
.consumerGroup("consumer-test")
.nameSrvAddr("127.0.0.1:9876")
.subscribe("test")
.orderlyProcessor(new OrderlyProcessor<MessageExt>() {
@Override
public ConsumeOrderlyStatus process(MessageExt messageBody) {
System.out.println("OrderlyProcessor, messageBody=" + messageBody);
return ConsumeOrderlyStatus.SUCCESS;
}
})
.start();
Thread.sleep(50000000);
}
@Test
public void orderlyProcessor() throws InterruptedException {
RMWrapper.with(RMConsumer.class)
.consumerGroup("consumer-test")
.nameSrvAddr("127.0.0.1:9876")
.subscribe("test")
.orderlyProcessor(new OrderlyProcessor<MessageBody>() {
@Override
public ConsumeOrderlyStatus process(MessageBody messageBody) {
System.out.println("OrderlyProcessor, messageBody=" + messageBody);
return ConsumeOrderlyStatus.SUCCESS;
}
})
.start();
Thread.sleep(50000000);
}
自定义消息序列化工具
- 用户也可以根据自己的喜好和业务要求定制自己的消息序列化工具，只需要实现MessageConverter接口
消息幂等
- 可通过设置消息keys实现
向多个集群发送消息（多生产者）
- 为每个producer设置不同的unitName或者instanceName
消息动态topic消费
- 创建多个consumer，每个consumer进行subscribe不同的topic
注意事项
- 单机（同一台服务器）只能创建一个消费者组，不管是集群消费还是广播消费（详见：MQClientInstance#registerConsumer）
- 单机如果多个消费同一个生产者发送的topic消息，需要创建不同的消费者组消费相应topic的消息