使用Java代码实现RocketMQ的生产与消费消息

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

// 1.创建一个DefaultMQProducer实例，指定生产者组名 
DefaultMQProducer producer = new DefaultMQProducer("my-producer-group1");  
// 2.设置NameServer的地址
producer.setNamesrvAddr("192.168.220.135:9876");    
// 3.启动生产者实例  
producer.start();   
// 4.使用for循环发送10条消息  
for (int i = 0; i < 10; i++) {  
    // 创建一条消息，指定Topic为"MyTopic1"，Tag标签为"TagA"，消息体为"hello rocketmq"加上循环变量的值，同时把字符串转换为字节数组  
    Message message = new Message("MyTopic1","TagA",("hello rocketmq"+i).getBytes(StandardCharsets.UTF_8));       
    // 5.发送消息并接收发送结果  
    SendResult sendResult = producer.send(message);  
    // 打印发送结果，包括消息ID、发送状态等信息  
    System.out.println(sendResult);  
}  
// 6.发送完所有消息后，关闭生产者实例，释放资源  
producer.shutdown();

// 1.和生产者一样，创建一个DefaultMQPushConsumer实例
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("my-consumer-group1");  
// 2.设置NameServer的地址，消费者通过这个地址与NameServer进行通信，来获取Broker的地址信息  
consumer.setNamesrvAddr("192.168.220.135:9876");  
// 订阅一个或多个Topic，以及Tag来过滤需要消费的消息。我们订阅了"MyTopic1"，使用"*"来匹配此Topic下的所有Tag  
consumer.subscribe("MyTopic1", "*");  
// 3.注册消息监听器，用于处理从Broker接收到的消息。使用MessageListenerConcurrently接口的实现，表示并行消费  
consumer.registerMessageListener(new MessageListenerConcurrently() {  
    @Override  
    // 4.当收到消息时，方法会被调用。
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {  
        // 5.遍历消息列表，并打印每条消息的内容（注意：这里直接打印msg对象不会得到预期的消息内容字符串）  
        for (MessageExt msg : msgs) {  
            // 所以我们打印msg.getBody()的内容，为了保留消息原样  
            System.out.println("已收到消息" + msg);  
        }  
        // 6.返回消费状态，这里表示消息已成功消费  
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;  
    }  
});  
// 7.启动消费者实例  
consumer.start();  
// 8.打印日志消费者已启动  
System.out.println("消费者已启动");