首页 > 代码库 > RocketMq入门之消息发送和接收

RocketMq入门之消息发送和接收

一、消息产生、发送

 1 public class Producer {
 2 public static void main(String[] args) throws MQClientException {
 3   DefaultMQProducer producer = new DefaultMQProducer("rmq-group");
 4   producer.setNamesrvAddr("172.18.4.114:9876");
 5   producer.setInstanceName("producer");
 6   producer.start();
 7   try {
 8     for (int i = 0; i < 10; i++) {
 9     Thread.sleep(5000); //每5秒发送一次MQ
10     Message msg = new Message("TopicA-test",// topic
11       "TagA",// tag
12       (new Date() + " Hello RocketMQ ,QuickStart" + i)
13       .getBytes()// body
14       );
15     SendResult sendResult = producer.send(msg);
16     }
17   } catch (Exception e) {
18     e.printStackTrace();
19   }
20   producer.shutdown();
21   }
22 }

二、消息接收、消费

 1 import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
 2 import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
 3 import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
 4 import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
 5 import com.alibaba.rocketmq.client.exception.MQClientException;
 6 import com.alibaba.rocketmq.common.message.MessageExt;
 7 
 8 import java.util.List;
 9 
10 
11 public class Consumer {
12     public static void main(String[] args) throws MQClientException {
13         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("rmq-group");
14 
15         consumer.setNamesrvAddr("172.18.4.114:9876");
16         consumer.setInstanceName("consumer");
17         consumer.subscribe("TopicA-test", "TagA");
18 
19         consumer.registerMessageListener(new MessageListenerConcurrently() {
20                 @Override
21                 public ConsumeConcurrentlyStatus consumeMessage(
22                     List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
23                     for (MessageExt msg : msgs) {
24                         System.out.println(new String(msg.getTopic()));
25                         System.out.println(new String(msg.getTags()));
26                         System.out.println("=== " + new String(msg.getBody()));
27                     }
28 
29                     return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
30                 }
31             });
32         consumer.start();
33         System.out.println("Consumer Started.");
34     }
35 }

 



RocketMq入门之消息发送和接收