<!--MQ-->
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.4.0</version>
        </dependency>

//發(fā)送同步消息
public class RocketMQSendTest1 {
    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        //1.創(chuàng)建消息生產(chǎn)者，指定生產(chǎn)者所屬的組名
        DefaultMQProducer producer = new DefaultMQProducer("myproducer-group");
        //2.指定Nameserver地址
        producer.setNamesrvAddr("127.0.0.1:9876");
        //3.啟動(dòng)生產(chǎn)者
        producer.start();
        for (int i = 0; i < 10; i++) {
            Message message = new Message("myTopic", "myTag", ("十行代碼九個(gè)錯(cuò)誤八個(gè)警告竟敢說七日精通六天學(xué)會(huì)五湖四海也不見如此三心二意之程序簡(jiǎn)直一等下流" + i).getBytes());
            //發(fā)送消息
            SendResult sendResult = producer.send(message);
            System.out.println(sendResult);
        }

        //關(guān)閉生產(chǎn)者
        producer.shutdown();
    }
}

//接收消息
public class RocketMQReceiveTest1 {
    public static void main(String[] args) throws MQClientException {
        //創(chuàng)建消息消費(fèi)者。指定消費(fèi)者所屬的組名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("myconsumer-group");
        //指定Nameserver地址
        consumer.setNamesrvAddr("127.0.0.1:9876");
        //指定消費(fèi)者訂閱的主題和標(biāo)簽
        consumer.subscribe("myTopic", "*");

        //設(shè)置回調(diào)函數(shù)，編寫處理消息的方法
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                System.out.println("獲取到的消費(fèi)數(shù)據(jù)：" + list);
                System.out.println(new String(list.get(0).getBody()));
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        //啟動(dòng)消息消費(fèi)者
        consumer.start();
        System.out.println("Consumer Starting.");
    }
}

//發(fā)送異步消息
//異步消息比較浪費(fèi)性能，經(jīng)常會(huì)失敗，所以多發(fā)送幾次并且讓線程休眠幾秒
public class RocketMQSendTest2 {

    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException {
        //1.創(chuàng)建消息生產(chǎn)者，指定生產(chǎn)者所屬的組名
        DefaultMQProducer producer = new DefaultMQProducer("myproducer-group");
        //2.設(shè)置NameServer地址
        producer.setNamesrvAddr("127.0.0.1:9876");
        //3.啟動(dòng)生產(chǎn)者
        producer.start();
        for (int i = 0; i < 10; i++) {
            //4.創(chuàng)建消息對(duì)象，制定主題、標(biāo)簽、消息體
            Message message = new Message("myTopic", "myTag2", ("十行代碼九個(gè)錯(cuò)誤八個(gè)警告竟敢說七日精通六天學(xué)會(huì)五湖四海也不見如此三心二意之程序簡(jiǎn)直一等下流").getBytes());
            //5.發(fā)送消息
            producer.send(message, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.println("發(fā)送成功：" + sendResult);
                }

                @Override
                public void onException(Throwable e) {
                    System.out.println("發(fā)送異常：" + e);
                }
            });

            //休眠
            TimeUnit.SECONDS.sleep(3);
        }

        //6.關(guān)閉生產(chǎn)者
        producer.shutdown();

    }
}

public class RocketMQReceiveTest2 {
    public static void main(String[] args) throws MQClientException {
        //1.創(chuàng)建消息消費(fèi)者，指定消費(fèi)者所屬的組名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("myconsumer-group");

        //2.指定Nameserver地址
        consumer.setNamesrvAddr("127.0.0.1:9876");
        //3.指定消費(fèi)者訂閱的主題和標(biāo)簽
        consumer.subscribe("myTopic", "*");
        consumer.setMessageModel(MessageModel.BROADCASTING);
        //4.設(shè)置回調(diào)函數(shù)，編寫處理請(qǐng)求的方法
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

                System.out.println(new String(list.get(0).getBody()));
                //返回消費(fèi)狀態(tài)
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.啟動(dòng)消息消費(fèi)者
        consumer.start();
        System.out.println("Consumer Starting.");

    }
}

//單行發(fā)送消息
public class RocketMQSendTest3 {
    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        DefaultMQProducer producer = new DefaultMQProducer("myproducer-group");
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.start();
        for (int i = 0; i < 10; i++) {
            Message message = new Message("myTopic", "myTag3", ("蕪湖").getBytes());
            producer.sendOneway(message);

            TimeUnit.SECONDS.sleep(3);
        }
        producer.shutdown();
    }

}