SpringBoot2.0快速集成RocketMQ（三）

参考：https://github.com/apache/rocketmq-spring/blob/master/README_zh_CN.md

RocketMQ 单机搭建以及使用：https://blog.csdn.net/qq_31289187/article/details/88351235

1、pom.xml

4.0.0

org.springframework.boot spring-boot-starter-parent 2.1.3.RELEASE com.cn.dl springboot-rocketmq 0.0.1-SNAPSHOT springboot-rocketmq springboot-rocketmq

1.8 org.springframework.boot spring-boot-starter org.apache.rocketmq rocketmq-spring-boot-starter 2.0.1 com.alibaba.rocketmq rocketmq-client 3.5.8 org.projectlombok lombok 1.18.4 com.alibaba fastjson 1.2.47 org.springframework.boot spring-boot-starter-test test

org.springframework.boot spring-boot-maven-plugin

2、application.properties

#nameServer rocketmq.name-server=127.0.0.1:9876 #生产者组 rocketmq.producer.group=user_producer_group

3、UserMsg

package com.cn.dl; import lombok.AllArgsConstructor; import lombok.Builder; import lombok.Data; import lombok.NoArgsConstructor; import java.io.Serializable; /** * Created by yanshao on 2019/3/29. */ @Data @Builder @NoArgsConstructor @AllArgsConstructor public class UserMsg implements Serializable { private long userId; private String name; private int age; private String address; }

4、Constant：配置

package com.cn.dl; /** * Created by yanshao on 2019/3/29. */ public interface Constant { String TOPIC = "user_msg"; String PRODUCER_GROUP = "user_producer_group"; String CONSUMER_GROUP = "user_consumer_group"; }

5、MsgProducer

package com.cn.dl; import com.alibaba.fastjson.JSON; import org.apache.rocketmq.spring.core.RocketMQTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; /** * Created by yanshao on 2019/3/29. */ @Component public class MsgProducer { @Autowired private RocketMQTemplate rocketMQTemplate; public void sendUserMsg(UserMsg userMsg){ rocketMQTemplate.syncSend(Constant.TOPIC, JSON.toJSONString(userMsg)); } }

6、MsgConsumer

package com.cn.dl; import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; import org.apache.rocketmq.common.message.MessageExt; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQListener; import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener; import org.springframework.stereotype.Service; import java.io.UnsupportedEncodingException; /** * Created by yanshao on 2019/3/29. */ @Service @RocketMQMessageListener( cOnsumerGroup= Constant.CONSUMER_GROUP, topic = Constant.TOPIC ) public class MsgConsumer implements RocketMQListener,RocketMQPushConsumerLifecycleListener { @Override public void onMessage(MessageExt messageExt) { String msg = null; try { msg = new String(messageExt.getBody(),"utf-8"); } catch (UnsupportedEncodingException e) { e.printStackTrace(); } System.out.println("MsgConsumer >>> " + msg +", msgId = " + messageExt.getMsgId()); } @Override public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) { } }

7、RocketMQTest

package com.cn.dl; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringRunner; import java.util.concurrent.TimeUnit; /** * Created by yanshao on 2019/3/29. */ @RunWith(SpringRunner.class) @SpringBootTest public class RocketMQTest { @Autowired private MsgProducer msgProducer; @Test public void rocketMqTest() throws InterruptedException { int i = 0; while (i <10){ UserMsg userMsg = UserMsg.builder() .userId(10000 + i) .name("yanshao" + i) .age(10 + i) .address("shanghai") .build(); i++; msgProducer.sendUserMsg(userMsg); TimeUnit.SECONDS.sleep(5); } } }

MsgConsumer >>> {"address":"shanghai","age":10,"name":"yanshao0","userId":10000}, msgId = 0200011D5EF418B4AAC2935F143C0000 MsgConsumer >>> {"address":"shanghai","age":11,"name":"yanshao1","userId":10001}, msgId = 0200011D5EF418B4AAC2935F27D00001 MsgConsumer >>> {"address":"shanghai","age":12,"name":"yanshao2","userId":10002}, msgId = 0200011D5EF418B4AAC2935F3B620002 MsgConsumer >>> {"address":"shanghai","age":13,"name":"yanshao3","userId":10003}, msgId = 0200011D5EF418B4AAC2935F4EF50003 MsgConsumer >>> {"address":"shanghai","age":14,"name":"yanshao4","userId":10004}, msgId = 0200011D5EF418B4AAC2935F62870004 MsgConsumer >>> {"address":"shanghai","age":15,"name":"yanshao5","userId":10005}, msgId = 0200011D5EF418B4AAC2935F761A0005 MsgConsumer >>> {"address":"shanghai","age":16,"name":"yanshao6","userId":10006}, msgId = 0200011D5EF418B4AAC2935F89AA0006 MsgConsumer >>> {"address":"shanghai","age":17,"name":"yanshao7","userId":10007}, msgId = 0200011D5EF418B4AAC2935F9D3B0007 MsgConsumer >>> {"address":"shanghai","age":18,"name":"yanshao8","userId":10008}, msgId = 0200011D5EF418B4AAC2935FB0CB0008 MsgConsumer >>> {"address":"shanghai","age":19,"name":"yanshao9","userId":10009}, msgId = 0200011D5EF418B4AAC2935FC45D0009

再启一个消费者，订阅user_msg

8、MsgConsumer2

package com.cn.dl; import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; import org.apache.rocketmq.common.message.MessageExt; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQListener; import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener; import org.springframework.stereotype.Service; import java.io.UnsupportedEncodingException; /** * Created by yanshao on 2019/3/29. */ @Service @RocketMQMessageListener( #记得修改consumer_group cOnsumerGroup= Constant.CONSUMER_GROUP, topic = Constant.TOPIC ) public class MsgConsumer2 implements RocketMQListener,RocketMQPushConsumerLifecycleListener { @Override public void onMessage(MessageExt messageExt) { String msg = null; try { msg = new String(messageExt.getBody(),"utf-8"); } catch (UnsupportedEncodingException e) { e.printStackTrace(); } System.out.println("MsgConsumer2 消费者2 >>> " + msg +", msgId = " + messageExt.getMsgId()); } @Override public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) { } }

Caused by: java.lang.IllegalStateException: Failed to start RocketMQ push consumer at org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer.start(DefaultRocketMQListenerContainer.java:231) at org.apache.rocketmq.spring.autoconfigure.ListenerContainerConfiguration.registerContainer(ListenerContainerConfiguration.java:100) ... 36 more Caused by: org.apache.rocketmq.client.exception.MQClientException: The consumer group[user_consumer_group] has been created before, specify another name please. See http://rocketmq.apache.org/docs/faq/ for further details. at org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl.start(DefaultMQPushConsumerImpl.java:613) at org.apache.rocketmq.client.consumer.DefaultMQPushConsumer.start(DefaultMQPushConsumer.java:520) at org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer.start(DefaultRocketMQListenerContainer.java:229) ... 37 more

       rocketmq不允许一个消费者同时订阅多个topic或者同时做多件事情，也就是消费者只能订阅一个topic、做一件事情，这跟kafka有很大的不同。重新定义一个消费者

MsgConsumer >>> {"address":"shanghai","age":10,"name":"yanshao0","userId":10000}, msgId = 0200011D3C2018B4AAC2937D5AFB0000 MsgConsumer2 消费者2 >>> {"address":"shanghai","age":11,"name":"yanshao1","userId":10001}, msgId = 0200011D3C2018B4AAC2937D6E8D0001 MsgConsumer >>> {"address":"shanghai","age":11,"name":"yanshao1","userId":10001}, msgId = 0200011D3C2018B4AAC2937D6E8D0001 MsgConsumer2 消费者2 >>> {"address":"shanghai","age":12,"name":"yanshao2","userId":10002}, msgId = 0200011D3C2018B4AAC2937D82240002 MsgConsumer >>> {"address":"shanghai","age":12,"name":"yanshao2","userId":10002}, msgId = 0200011D3C2018B4AAC2937D82240002 MsgConsumer2 消费者2 >>> {"address":"shanghai","age":13,"name":"yanshao3","userId":10003}, msgId = 0200011D3C2018B4AAC2937D95B70003 MsgConsumer >>> {"address":"shanghai","age":13,"name":"yanshao3","userId":10003}, msgId = 0200011D3C2018B4AAC2937D95B70003 MsgConsumer2 消费者2 >>> {"address":"shanghai","age":14,"name":"yanshao4","userId":10004}, msgId = 0200011D3C2018B4AAC2937DA9430004 MsgConsumer >>> {"address":"shanghai","age":14,"name":"yanshao4","userId":10004}, msgId = 0200011D3C2018B4AAC2937DA9430004

rocketmq消费失败，默认重试16次，时间间隔：10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

如果捕获处理逻辑代码，并没有抛出异常，rocketmq不会重试消费的

MsgConsumer >>> {"address":"shanghai","age":10,"name":"yanshao0","userId":10000}, msgId = 0200011D61E418B4AAC2938398610000 at com.cn.dl.MsgConsumer.onMessage(MsgConsumer.java:32) at com.cn.dl.MsgConsumer.onMessage(MsgConsumer.java:15) at org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer$DefaultMessageListenerConcurrently.consumeMessage(DefaultRocketMQListenerContainer.java:295) at org.apache.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService$ConsumeRequest.run(ConsumeMessageConcurrentlyService.java:417) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) java.lang.ArithmeticException: / by zero at com.cn.dl.MsgConsumer.onMessage(MsgConsumer.java:32) at com.cn.dl.MsgConsumer.onMessage(MsgConsumer.java:15) at org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer$DefaultMessageListenerConcurrently.consumeMessage(DefaultRocketMQListenerContainer.java:295) at org.apache.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService$ConsumeRequest.run(ConsumeMessageConcurrentlyService.java:417) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) java.lang.ArithmeticException: / by zero MsgConsumer >>> {"address":"shanghai","age":11,"name":"yanshao1","userId":10001}, msgId = 0200011D61E418B4AAC29383ABF00001 at com.cn.dl.MsgConsumer.onMessage(MsgConsumer.java:32) at com.cn.dl.MsgConsumer.onMessage(MsgConsumer.java:15) at org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer$DefaultMessageListenerConcurrently.consumeMessage(DefaultRocketMQListenerContainer.java:295) at org.apache.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService$ConsumeRequest.run(ConsumeMessageConcurrentlyService.java:417) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) java.lang.ArithmeticException: / by zero at com.cn.dl.MsgConsumer.onMessage(MsgConsumer.java:32) at com.cn.dl.MsgConsumer.onMessage(MsgConsumer.java:15) at org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer$DefaultMessageListenerConcurrently.consumeMessage(DefaultRocketMQListenerContainer.java:295) at org.apache.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService$ConsumeRequest.run(ConsumeMessageConcurrentlyService.java:417) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) MsgConsumer >>> {"address":"shanghai","age":12,"name":"yanshao2","userId":10002}, msgId = 0200011D61E418B4AAC29383BF7C0002 MsgConsumer >>> {"address":"shanghai","age":13,"name":"yanshao3","userId":10003}, msgId = 0200011D61E418B4AAC29383D30C0003java.lang.ArithmeticException: / by zero at com.cn.dl.MsgConsumer.onMessage(MsgConsumer.java:32) at com.cn.dl.MsgConsumer.onMessage(MsgConsumer.java:15) at org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer$DefaultMessageListenerConcurrently.consumeMessage(DefaultRocketMQListenerContainer.java:295) at

如果抛出异常，消费失败的，会重试

      如果消费失败，也不重试，在void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer)设置，也可以设置offset等等。

每天进步一点点！！！