一、生产者

　　使用RocketMQ以三种方式发送消息：可靠的同步，可靠的异步和单向传输。

　　（1）同步发送消息（可靠的同步传输，适用于重要的短信通知等）

public class SyncProducer {    public static void main(String[] args) throws Exception {        //Instantiate with a producer group name.        DefaultMQProducer producer = new            DefaultMQProducer("please_rename_unique_group_name");        // Specify name server addresses.        producer.setNamesrvAddr("localhost:9876");        //Launch the instance.        producer.start();        for (int i = 0; i < 100; i++) {            //Create a message instance, specifying topic, tag and message body.            Message msg = new Message("TopicTest" /* Topic */,                "TagA" /* Tag */,                ("Hello RocketMQ " +                    i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */            );            //Call send message to deliver message to one of brokers.            SendResult sendResult = producer.send(msg);            System.out.printf("%s%n", sendResult);        }        //Shut down once the producer instance is not longer in use.        producer.shutdown();    }}

 　　(2)异步传输通常用于响应时间敏感的业务场景。

public class AsyncProducer {    public static void main(String[] args) throws Exception {        //Instantiate with a producer group name.        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");        // Specify name server addresses.        producer.setNamesrvAddr("localhost:9876");        //Launch the instance.        producer.start();        producer.setRetryTimesWhenSendAsyncFailed(0);        for (int i = 0; i < 100; i++) {                final int index = i;                //Create a message instance, specifying topic, tag and message body.                Message msg = new Message("TopicTest",                    "TagA",                    "OrderID188",                    "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));                producer.send(msg, new SendCallback() {                    @Override                    public void onSuccess(SendResult sendResult) {                        System.out.printf("%-10d OK %s %n", index,                            sendResult.getMsgId());                    }                    @Override                    public void onException(Throwable e) {                        System.out.printf("%-10d Exception %s %n", index, e);                        e.printStackTrace();                    }                });        }        //Shut down once the producer instance is not longer in use.        producer.shutdown();    }}

　　(3)以单向模式发送消息(单向传输用于需要中等可靠性的情况，例如日志收集。)

public class OnewayProducer {    public static void main(String[] args) throws Exception{        //Instantiate with a producer group name.        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");        // Specify name server addresses.        producer.setNamesrvAddr("localhost:9876");        //Launch the instance.        producer.start();        for (int i = 0; i < 100; i++) {            //Create a message instance, specifying topic, tag and message body.            Message msg = new Message("TopicTest" /* Topic */,                "TagA" /* Tag */,                ("Hello RocketMQ " +                    i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */            );            //Call send message to deliver message to one of brokers.            producer.sendOneway(msg);        }        //Shut down once the producer instance is not longer in use.        producer.shutdown();    }}

二、消费者

　　

public class Consumer {    public static void main(String[] args) throws InterruptedException, MQClientException {        // Instantiate with specified consumer group name.        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");                 // Specify name server addresses.        consumer.setNamesrvAddr("localhost:9876");                // Subscribe one more more topics to consume.        consumer.subscribe("TopicTest", "*");        // Register callback to execute on arrival of messages fetched from brokers.        consumer.registerMessageListener(new MessageListenerConcurrently() {            @Override            public ConsumeConcurrentlyStatus consumeMessage(List
     
       msgs,                ConsumeConcurrentlyContext context) {                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;            }        });        //Launch the consumer instance.        consumer.start();        System.out.printf("Consumer Started.%n");    }}