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

public class Producer {
    public static void main(String[] args) throws Exception {
        //創(chuàng)建一個生產(chǎn)者，指定生產(chǎn)者組為jihaiProducer
        DefaultMQProducer producer = new DefaultMQProducer("jihaiProducer");
        // 指定NameServer的地址
        producer.setNamesrvAddr("localhost:9876");
        // 第一次發(fā)送可能會超時，設(shè)置的比較大
        producer.setSendMsgTimeout(60000);
        // 啟動生產(chǎn)者
        producer.start();
        // 創(chuàng)建一條消息
        // topic為 jihaishibei
        // 消息內(nèi)容為 技海拾貝的消息1
        // tags 為 TagA
        Message msg = new Message("jihaishibei", "TagA", "技海拾貝的消息1 ".getBytes(RemotingHelper.DEFAULT_CHARSET));
        // 發(fā)送消息并得到消息的發(fā)送結(jié)果，然后打印
        SendResult sendResult = producer.send(msg);
        System.out.printf("%s%n", sendResult);
        // 關(guān)閉生產(chǎn)者
        producer.shutdown();
    }
}

public class Consumer {
    public static void main(String[] args) throws InterruptedException, MQClientException {
        // 通過push模式消費消息，指定消費者組
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("jihaiConsumer");
        // 指定NameServer的地址
        consumer.setNamesrvAddr("localhost:9876");
        // 訂閱這個topic下的所有的消息
        consumer.subscribe("jihaishibei", "*");
        // 注冊一個消費的監(jiān)聽器，當有消息的時候，會回調(diào)這個監(jiān)聽器來消費消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.printf("消費消息:%s", new String(msg.getBody()) + "\n");
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 啟動消費者
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}

spring:
  application:
    name: rocket-mq-demo
rocketmq:
  name-server: 127.0.0.1:9876
  producer:
    group: rocket-mq-demo-producer
    send-message-timeout: 10000
  comsumer:
    group: rocket-mq-demo-comsumer
    send-message-timeout: 10000

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

import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;
/**
 * messageModel=MessageModel.CLUSTERING
 * 監(jiān)聽模式，有消息就會消費
 */
@Service
@RocketMQMessageListener(topic = "jihaishibei-topic", consumerGroup = "rocket-mq-demo-comsumer", messageModel = MessageModel.CLUSTERING)
public class RocketMQConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String s) {
        System.out.printf("收到消息: %s\n", s);
    }
}

import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
@Service
public class RocketMQProducer {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    private final String topic = "jihaishibei-topic";
    // 1.同步發(fā)送消息
    // 同步發(fā)送是指發(fā)送方發(fā)送一條消息后，會等待服務(wù)器返回確認信息后再進行后續(xù)操作。這種方式適用于需要可靠性保證的場景。
    public void createAndSend(String message){
        rocketMQTemplate.convertAndSend(topic, message);
        System.out.printf("同步發(fā)送結(jié)果: %s\n", message);
    }
    // 1.同步發(fā)送消息
    // 同步發(fā)送是指發(fā)送方發(fā)送一條消息后，會等待服務(wù)器返回確認信息后再進行后續(xù)操作。這種方式適用于需要可靠性保證的場景。
    public void sendSyncMessage(String message){
        SendResult sendResult = rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(message).build());
        System.out.println(sendResult.getMsgId());
        System.out.printf("同步發(fā)送結(jié)果: %s\n", message);
    }
    // 2.異步發(fā)送消息
    // 異步發(fā)送是指發(fā)送方發(fā)送消息后，不等待服務(wù)器返回確認信息，而是通過回調(diào)接口處理返回結(jié)果。這種方式適用于對響應(yīng)時間要求較高的場景。
    public void sendAsyncMessage(String message){
        rocketMQTemplate.asyncSend(topic, MessageBuilder.withPayload(message).build(), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.printf("異步發(fā)送成功: %s\n", sendResult);
            }
            @Override
            public void onException(Throwable throwable) {
                System.out.printf("異步發(fā)送失敗: %s\n", throwable.getMessage());
            }
        });
    }
    // 3.單向發(fā)送消息
    // 單向發(fā)送是指發(fā)送方只負責發(fā)送消息，不關(guān)心服務(wù)器的響應(yīng)。該方式適用于對可靠性要求不高的場景，如日志收集。
    public void sendOneWayMessage(String message){
        rocketMQTemplate.sendOneWay(topic, MessageBuilder.withPayload(message).build());
        System.out.println("單向消息發(fā)送成功");
    }
}

@RequestMapping("api")
@RestController
public class RocketController {
    @Autowired
    private RocketMQProducer rocketMQProducer;
    @GetMapping("/createAndSend")
    public String createAndSend(@RequestParam String message) {
        rocketMQProducer.createAndSend(message);
        return "同步消息發(fā)送成功";
    }
    @GetMapping("/sendSync")
    public String sendSync(@RequestParam String message) {
        rocketMQProducer.sendSyncMessage(message);
        return "同步消息發(fā)送成功";
    }
    @GetMapping("/sendAsync")
    public String sendAsync(@RequestParam String message) {
        rocketMQProducer.sendAsyncMessage(message);
        return "異步消息發(fā)送中";
    }
    @GetMapping("/sendOneWay")
    public String sendOneWay(@RequestParam String message) {
        rocketMQProducer.sendOneWayMessage(message);
        return "單向消息發(fā)送成功";
    }
}