RocketMQ安装部署及整合Springboot

JAVA_OPT="${JAVA_OPT} -server -Xms1g -Xmx1g -Xmn256m"
Xms:是指设定程序启动时占用内存大小。一般来讲，大点，程序会启动的快一点，但是也可能会导致机器暂时间变慢。
Xmx:是指设定程序运行期间最大可占用的内存大小。如果程序运行需要占用更多的内存，超出了这个设置值，就会抛出OutOfMemory异常。
xmn:年轻代的heap大小，一般设置为Xmx的3、4分之一

<dependency>
  <groupId>org.apache.rocketmq</groupId>
  <artifactId>rocketmq-client</artifactId>
  <version>4.5.</version>
</dependency>

public class RocketMqProducer {
public static void main(String[] args) throws MQClientException, InterruptedException {
/*
     *生产者组，简单来说就是多个发送同一类消息的生产者称之为一个生产者组
*rocketmq支持事务消息，在发送事务消息时，如果事务消息异常（producer挂了），broker端会来回查
*事务的状态，这个时候会根据group名称来查找对应的producer来执行相应的回查逻辑。相当于实现了producer的高可用
     */
DefaultMQProducer producer = new DefaultMQProducer("unique_producer_group_name");
producer.setDefaultTopicQueueNums(3);//设置默认的queue数量
//指定namesrv服务地址，获取broker相关信息
producer.setNamesrvAddr("192.168.1.101:9876");
producer.start();
for (int i = ; i < ; i++) {
try {
//创建一个消息实例，指定指定topic、tag、消息内容
Message msg = new Message("testTopic", "testTag",
("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */);
//发送消息并且获取发送结果
SendResult sendResult = producer.send(msg);
System.out.printf("%s%n", sendResult);
} catch (Exception e) {
e.printStackTrace();
Thread.sleep();
}
}
producer.shutdown();
}
}

public class RocketMqConsumer {
public static void main(String[] args) throws MQClientException {
//消费者的组名，这个和kafka是一样,这里需要注意的是，
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("unique_consumer_group_name");
//指定NameServer地址，多个地址以 ; 隔开
consumer.setNamesrvAddr("192.168.1.101:9876");
//设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
//如果非第一次启动，那么按照上次消费的位置继续消费
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
//订阅PushTopic下Tag为push的消息
consumer.subscribe("testTopic", "*"); //*表示不过滤，可以通过tag来过滤，比如:”tagA”
/*
     * 注册消息监听回调这里有两种监听，MessageListenerConcurrently以及MessageListenerOrderly
* 前者是普通监听，后者是顺序监听。这块在后续单独说明
     */
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt>
msgs,
ConsumeConcurrentlyContext context) {
System.out.printf("%s Receive New Messages: %s %n",
Thread.currentThread().getName(), msgs);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//返回消息消费状态
}
});
consumer.start();
System.out.printf("Consumer Started.%n"); }

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.5.</version>
</dependency>

rocketmq:
# 生产者配置
producer:
isOnOff: on
# 发送同一类消息的设置为同一个group，保证唯一
groupName: unique_producer_group_name
# 服务地址
namesrvAddr: 192.168.1.101:
# 消息最大长度 默认1024*(4M)
maxMessageSize:
# 发送消息超时时间,默认3000
sendMsgTimeout:
# 发送消息失败重试次数，默认2
retryTimesWhenSendFailed:
# 消费者配置
consumer:
isOnOff: on
# 官方建议：确保同一组中的每个消费者订阅相同的主题。
groupName: unique_consumer_group_name
# 服务地址
namesrvAddr: 192.168.1.101:
# 接收该 Topic 下所有 Tag
topics: testTopic~*;
consumeThreadMin:
consumeThreadMax:
# 设置一次消费消息的条数，默认为1条
consumeMessageBatchMaxSize: # 配置 Group Topic Tag
plat:
plat-group: unique_group_name
plat-topic: testTopic
plat-tag: testTag

@Configuration
public class ProducerConfig {
private static final Logger LOG = LoggerFactory.getLogger(ProducerConfig.class) ;
@Value("${rocketmq.producer.groupName}")
private String groupName;
@Value("${rocketmq.producer.namesrvAddr}")
private String namesrvAddr;
@Value("${rocketmq.producer.maxMessageSize}")
private Integer maxMessageSize ;
@Value("${rocketmq.producer.sendMsgTimeout}")
private Integer sendMsgTimeout;
@Value("${rocketmq.producer.retryTimesWhenSendFailed}")
private Integer retryTimesWhenSendFailed;
@Bean
public DefaultMQProducer defaultMQProducer() {
DefaultMQProducer producer;
producer = new DefaultMQProducer(this.groupName);
producer.setNamesrvAddr(this.namesrvAddr);
//如果需要同一个jvm中不同的producer往不同的mq集群发送消息，需要设置不同的instanceName
if(this.maxMessageSize!=null){
producer.setMaxMessageSize(this.maxMessageSize);
}
if(this.sendMsgTimeout!=null){
producer.setSendMsgTimeout(this.sendMsgTimeout);
}
//如果发送消息失败，设置重试次数，默认为2次
if(this.retryTimesWhenSendFailed!=null){
producer.setRetryTimesWhenSendFailed(this.retryTimesWhenSendFailed);
}
try {
producer.start();
} catch (MQClientException e) {
e.printStackTrace();
}
return producer;
}
}

@Configuration
public class ConsumerConfig {
private static final Logger LOG = LoggerFactory.getLogger(ConsumerConfig.class) ;
@Value("${rocketmq.consumer.namesrvAddr}")
private String namesrvAddr;
@Value("${rocketmq.consumer.groupName}")
private String groupName;
@Value("${rocketmq.consumer.consumeThreadMin}")
private int consumeThreadMin;
@Value("${rocketmq.consumer.consumeThreadMax}")
private int consumeThreadMax;
@Value("${rocketmq.consumer.topics}")
private String topics;
@Value("${rocketmq.consumer.consumeMessageBatchMaxSize}")
private int consumeMessageBatchMaxSize;
@Resource
private RocketMsgListener msgListener;
@Bean
public DefaultMQPushConsumer defaultMQPushConsumer(){
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
consumer.setNamesrvAddr(namesrvAddr);
consumer.setConsumeThreadMin(consumeThreadMin);
consumer.setConsumeThreadMax(consumeThreadMax);
consumer.registerMessageListener(msgListener);
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
try {
String[] topicTagsArr = topics.split(";");
for (String topicTags : topicTagsArr) {
String[] topicTag = topicTags.split("~");
consumer.subscribe(topicTag[],topicTag[]);
}
consumer.start();
}catch (MQClientException e){
e.printStackTrace();
}
return consumer;
}
}

@Component
public class RocketMsgListener implements MessageListenerConcurrently {
private static final Logger LOG = LoggerFactory.getLogger(RocketMsgListener.class) ;
@Resource
private ParamConfigService paramConfigService ; @Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
if (CollectionUtils.isEmpty(list)){
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
MessageExt messageExt = list.get();
LOG.info("接受到的消息为："+new String(messageExt.getBody()));
int reConsume = messageExt.getReconsumeTimes();
// 消息已经重试了3次，如果不需要再次消费，则返回成功
if(reConsume ==){
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
if(messageExt.getTopic().equals(paramConfigService.platTopic)){
String tags = messageExt.getTags() ;
switch (tags){
case "testTag":
LOG.info("匹配到testTag"+tags);
break ;
default:
LOG.info("未匹配到Tag == >>"+tags);
break;
}
}
// 消息消费成功
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
}

@Service
public class ParamConfigService { @Value("${plat.plat-group}")
public String platGroup ;
@Value("${plat.plat-topic}")
public String platTopic ;
@Value("${plat.plat-tag}")
public String accountTag ;
//省略 get set
}

@RestController
public class TestController { @Autowired
private DefaultMQProducer defaultMQProducer; @Autowired
private ParamConfigService paramConfigService; @RequestMapping(value = "/testStringQueue.json", method = {RequestMethod.GET})
public SendResult testStringQueue() {
// 可以不使用Config中的Group
defaultMQProducer.setProducerGroup(paramConfigService.platGroup);
SendResult sendResult = null;
String msgInfo = "rocketmq message 1";
try {
Message sendMsg = new Message(paramConfigService.platTopic,
paramConfigService.accountTag, msgInfo.getBytes());
sendResult = defaultMQProducer.send(sendMsg);
} catch (Exception e) {
e.printStackTrace();
}
return sendResult;
}
}