1、生產(chǎn)者
package com.example.demo.rocketmq;

/**

• @Description:
• @author: YuanTong-ZXY
• @Date: 2019/2/22 9:34
  */
  import org.apache.rocketmq.client.producer.DefaultMQProducer;
  import org.apache.rocketmq.client.producer.SendResult;
  import org.apache.rocketmq.common.message.Message;

public class Producer {
public static void main(String[] args){
DefaultMQProducer producer = new DefaultMQProducer("Producer");
producer.setNamesrvAddr("192.168.205.196:9876");
try {
producer.start();

        Message msg = new Message("PushTopic",
                "push",
                "1",
                "Just for test.".getBytes());

        SendResult result = producer.send(msg);
        System.out.println("id:" + result.getMsgId() +
                " result:" + result.getSendStatus());

        msg = new Message("PushTopic",
                "push",
                "2",
                "Just for test.zhangxinyu111".getBytes());

        result = producer.send(msg);
        System.out.println("id:" + result.getMsgId() +
                " result:" + result.getSendStatus());

        msg = new Message("PullTopic",
                "pull",
                "1",
                "Just for test.".getBytes());

        result = producer.send(msg);
        System.out.println("id:" + result.getMsgId() +
                " result:" + result.getSendStatus());
    } catch (Exception e) {
        e.printStackTrace();
    }finally{
        producer.shutdown();
    }
}

}

2、消費(fèi)者
package com.example.demo.rocketmq;

/**

• @Description:
• @author: YuanTong-ZXY
• @Date: 2019/2/22 9:42
  */

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class Consumer {
public static void main(String[] args){
DefaultMQPushConsumer consumer =
new DefaultMQPushConsumer("PushConsumer");
consumer.setNamesrvAddr("192.168.205.196:9876");
try {
//訂閱PushTopic下Tag為push的消息
consumer.subscribe("PushTopic", "push");
//程序第一次啟動(dòng)從消息隊(duì)列頭取數(shù)據(jù)
consumer.setConsumeFromWhere(
ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.registerMessageListener(
new MessageListenerConcurrently() {
public ConsumeConcurrentlyStatus consumeMessage(
List<MessageExt> list,
ConsumeConcurrentlyContext Context) {
Message msg = list.get(0);
System.out.println(msg.toString());
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
}
);
consumer.start();
} catch (Exception e) {
e.printStackTrace();
}
}
}