<dependency>
 <groupId>org.apache.rocketmq</groupId>
 <artifactId>rocketmq-spring-boot-starter</artifactId>
 <version>2.3.0</version>
</dependency>
<dependency>
 <groupId>com.alibaba</groupId>
 <artifactId>fastjson</artifactId>
 <version>2.0.32</version>
</dependency>

rocketmq:
 name-server: 192.168.68.133:9876
 producer:
 #продюсер Название группы должно быть уникальным в пределах приложения
 group: test-producer
 #Тайм-аут отправки сообщения По умолчанию 3000 мс
 send-message-timeout: 3000
 #Когда сообщение достигнет размера 4096 байт, оно будет сжато. по умолчанию 4096
 compress-message-body-threshold: 4096
 #Максимальный лимит сообщений, по умолчанию — 128 КБ
 max-message-size: 4194304
 #Количество попыток отправки сообщения о неудачной синхронизации
 retry-times-when-send-failed: 2
 #Стоит ли повторять попытки использования других брокеров в случае сбоя внутренней отправки. Этот параметр вступает в силу только при наличии нескольких брокеров.
 retry-next-server: true
 #Количество попыток неудачной отправки асинхронного сообщения
 retry-times-when-send-async-failed: 2

@Test
void sendMsg() {
 /**
 * Отправить сообщение синхронизации
 * destination тема назначения
 * payload информация
 */
 rocketMQTemplate.syncSend("TopicTest", «синхронная информация»);
}

@Component
@RocketMQMessageListener(topic = "TopicTest",consumerGroup = "ConsumerGroup-springboot")
public class RocketListener implements RocketMQListener<MessageExt> {

 /**
 * onMessage потребительский метод
 * @param messages Содержание сообщения
 */
 @Override
 public void onMessage(MessageExt messages) {
 //Если вы не сообщаете об ошибке, вы подписываетесь для получения информации.
 //Сообщение об ошибке означает повторную попытку. Если сообщение об ошибке все еще сообщается после трех повторных попыток, это означает, что потребление не удалось и оно будет повторно введено в очередь. (согласно конфигурации yml)
 System.out.println("Получено сообщение: "+new String(messages.getBody()));
 }
}

@Test
void asyncTest() {
 /**
 * Отправлять асинхронные сообщения
 * destination тема назначения
 * payload информация
 */
 rocketMQTemplate.asyncSend("TopicTest", "асинхронныйинформация", new SendCallback() {
 @Override
 public void onSuccess(SendResult sendResult) {
 System.out.println("Отправлено успешно");
 }

 @Override
 public void onException(Throwable throwable) {
 System.out.println("Отправка не удалась");
 }
 });
}

@Component
@RocketMQMessageListener(topic = "TopicTest",consumerGroup = "ConsumerGroup-springboot")
public class RocketListener implements RocketMQListener<MessageExt> {

 /**
 * onMessage потребительский метод
 * @param messages Содержание сообщения
 */
 @Override
 public void onMessage(MessageExt messages) {
 //Если вы не сообщаете об ошибке, вы подписываетесь для получения информации.
 //Сообщение об ошибке означает повторную попытку. Если сообщение об ошибке все еще сообщается после трех повторных попыток, это означает, что потребление не удалось и оно будет повторно введено в очередь. (согласно конфигурации yml)
 System.out.println("Получено сообщение: "+new String(messages.getBody()));
 }
}

@Test
void delayTest() {
 /**
 * Отправить отложенное сообщение
 * destination тема назначения
 * payload информация
 * timestamp Тайм-аут соединения
 * delayLevel Уровень задержки
 */
 Message<String> msg = MessageBuilder.withPayload("задерживатьинформация").build();
 rocketMQTemplate.syncSend("TopicTest", msg, 3000, 3);
}

@Component
@RocketMQMessageListener(topic = "TopicTest",consumerGroup = "ConsumerGroup-springboot")
public class RocketListener implements RocketMQListener<MessageExt> {

 /**
 * onMessage потребительский метод
 * @param messages Содержание сообщения
 */
 @Override
 public void onMessage(MessageExt messages) {
 //Если вы не сообщаете об ошибке, вы подписываетесь для получения информации.
 //Сообщение об ошибке означает повторную попытку. Если сообщение об ошибке все еще сообщается после трех повторных попыток, это означает, что потребление не удалось и оно будет повторно введено в очередь. (согласно конфигурации yml)
 System.out.println("Получено сообщение: "+new String(messages.getBody()));
 }
}

@Test
void OneWayTest() {
 /**
 * Отправить одностороннее сообщение
 * destination тема назначения
 * payload информация
 */
 rocketMQTemplate.sendOneWay("TopicTest", «односторонняя информация»);
}

@Component
@RocketMQMessageListener(topic = "TopicTest",consumerGroup = "ConsumerGroup-springboot")
public class RocketListener implements RocketMQListener<MessageExt> {

 /**
 * onMessage потребительский метод
 * @param messages Содержание сообщения
 */
 @Override
 public void onMessage(MessageExt messages) {
 //Если вы не сообщаете об ошибке, вы подписываетесь для получения информации.
 //Сообщение об ошибке означает повторную попытку. Если сообщение об ошибке все еще сообщается после трех повторных попыток, это означает, что потребление не удалось и оно будет повторно введено в очередь. (согласно конфигурации yml)
 System.out.println("Получено сообщение: "+new String(messages.getBody()));
 }
}

List<MessageM> messageMs = Arrays.asList(
 new MessageM("sn0001", 1, «Оформить заказ»),
 new MessageM("sn0001", 1, "Оплата"),
 new MessageM("sn0001", 1, "Доставка"),
 new MessageM("sn0002", 2, «Оформить заказ»),
 new MessageM("sn0002", 2, "Оплата"),
 new MessageM("sn0002", 2, "Доставка")
);


@Test
void orderlyTest() {
 /**

 * destination тема назначения
 * payload информация




 */
 for (MessageM messageM : messageMs) {
 rocketMQTemplate.syncSendOrderly("orderlyTest", JSON.toJSON(messageM), messageM.getSn());
 }
}

@Component
@RocketMQMessageListener(topic = "orderlyTest",consumerGroup = "orderly",consumeMode = ConsumeMode.ORDERLY)
public class orderlyListener implements RocketMQListener<MessageExt> {

 @Override
 public void onMessage(MessageExt messageExt) {
 MessageM messageM = JSON.parseObject(new String(messageExt.getBody()), MessageM.class);
 System.out.println(messageM);
 }
}

@Test
void ProducerTagTest(){
 rocketMQTemplate.syncSend("TagMQ:tagA","Информация tagA");
 rocketMQTemplate.syncSend("TagMQ:tagB","приноситьtagBизинформация");
}

@Component
@RocketMQMessageListener(topic = "TagMQ",
 consumerGroup = "TagMQGroup",
 selectorType = SelectorType.TAG, //режим фильтра тегов
 selectorExpression = "tagA || tagB"

)
public class MsgListenerTag implements RocketMQListener<MessageExt> {

 @Override
 public void onMessage(MessageExt messageExt) {
 System.out.println(new String( messageExt.getBody()));
 }
}

@Test
void keyTest(){
 String Key = UUID.randomUUID().toString();
 Message<String> msg = MessageBuilder
 .withPayload("приноситьkeyинформация").
 setHeader(RocketMQHeaders.KEYS, Key)
 .build();
 /**
 * приноситьKeyинформация
 */
 rocketMQTemplate.syncSend("ketTopic",msg);
}

@Component
@RocketMQMessageListener(topic = "ketTopic",consumerGroup = "ketConsumerGroup-springboot")
public class keyMQListener implements RocketMQListener<MessageExt> {

 /**
 * onMessage потребительский метод
 * @param messages Содержание сообщения
 */
 @Override
 public void onMessage(MessageExt messages) {
 //Если вы не сообщаете об ошибке, вы подписываетесь для получения информации.
 //Сообщение об ошибке означает повторную попытку. Если сообщение об ошибке все еще сообщается после трех повторных попыток, это означает, что потребление не удалось и оно будет повторно введено в очередь. (согласно конфигурации yml)
 System.out.println("Получено сообщение: "+new String(messages.getBody()));
 System.out.println("key:"+messages.getKeys());
 }
}