<dependencies>
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.8.0</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.13.2</version>
        </dependency>
</dependencies>

    private final static String nameServer = "127.0.0.1:9876";
    private final static String producerGroup = "my_group";
    private final static String topic = "topic-test";
@Test
    public void syncSend() {
        try {
            // 初始化一個(gè)producer并設(shè)置Producer group name
            DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
            // 設(shè)置NameServer地址
            producer.setNamesrvAddr(nameServer);
            // 啟動producer
            producer.start();
            // 創(chuàng)建一條消息，并指定topic、tag、body等信息，tag可以理解成標(biāo)簽，對消息進(jìn)行再歸類，RocketMQ可以在消費(fèi)端對tag進(jìn)行過濾
            Message msg = new Message(topic, "tagA", "Hello RocketMQ".getBytes(RemotingHelper.DEFAULT_CHARSET));
            // 利用producer進(jìn)行發(fā)送，并同步等待發(fā)送結(jié)果
            SendResult sendResult = producer.send(msg, 10000);
            System.out.printf("%s%n", sendResult);
            // 一旦producer不再使用，關(guān)閉producer
            producer.shutdown();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

@Test
    public void asyncSend() throws IOException {
        try {
            // 初始化一個(gè)producer并設(shè)置Producer group name
            DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
            // 設(shè)置NameServer地址
            producer.setNamesrvAddr(nameServer);
            // 啟動producer
            producer.start();
            // 創(chuàng)建一條消息，并指定topic、tag、body等信息，tag可以理解成標(biāo)簽，對消息進(jìn)行再歸類，RocketMQ可以在消費(fèi)端對tag進(jìn)行過濾
            Message msg = new Message(topic, "tagB", "Hello RocketMQ".getBytes(RemotingHelper.DEFAULT_CHARSET));
            // 異步發(fā)送消息, 發(fā)送結(jié)果通過callback返回給客戶端
            producer.send(msg, new SendCallback() {
                public void onSuccess(SendResult sendResult) {
                    System.out.printf("OK %s %n",
                            sendResult.getMsgId());
                }
                public void onException(Throwable e) {
                    System.out.printf("Exception %s %n", e);
                    e.printStackTrace();
                }
            },10000);
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.in.read();
    }

@Test
    public void onewaySend()  {
        try {
            // 初始化一個(gè)producer并設(shè)置Producer group name
            DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
            // 設(shè)置NameServer地址
            producer.setNamesrvAddr(nameServer);
            producer.setSendMsgTimeout(10000);
            // 啟動producer
            producer.start();
            // 創(chuàng)建一條消息，并指定topic、tag、body等信息，tag可以理解成標(biāo)簽，對消息進(jìn)行再歸類，RocketMQ可以在消費(fèi)端對tag進(jìn)行過濾
            Message msg = new Message(topic, "tagB", "Hello RocketMQ".getBytes(RemotingHelper.DEFAULT_CHARSET));
            // 異步發(fā)送消息, 發(fā)送結(jié)果通過callback返回給客戶端
            producer.sendOneway(msg);
            // 一旦producer不再使用，關(guān)閉producer
            //producer.shutdown();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

@Test
    public void orderSend() {
        try {
            // 初始化一個(gè)producer并設(shè)置Producer group name
            DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
            // 設(shè)置NameServer地址
            producer.setNamesrvAddr(nameServer);
            producer.setSendMsgTimeout(10000);
            // 啟動producer
            producer.start();
            String[] tags = new String[]{"TagA", "TagB", "TagC"};
            for (int i = 0; i < 10; i++) {
                int orderId = i % 10;
                Message msg = new Message(topic, tags[i % tags.length], "KEY" + i, ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        Integer id = (Integer) arg;
                        int index = id % mqs.size();
                        return mqs.get(index);
                    }
                }, orderId);
                System.out.printf("%s%n", sendResult);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

@Test
    public void scheduledSend() {
        try {
            // 初始化一個(gè)producer并設(shè)置Producer group name
            DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
            // 設(shè)置NameServer地址
            producer.setNamesrvAddr(nameServer);
            // 啟動producer
            producer.start();
            // 創(chuàng)建一條消息，并指定topic、tag、body等信息，tag可以理解成標(biāo)簽，對消息進(jìn)行再歸類，RocketMQ可以在消費(fèi)端對tag進(jìn)行過濾
            Message msg = new Message(topic, "tagA", "Hello RocketMQ".getBytes(RemotingHelper.DEFAULT_CHARSET));
            // 消息延遲等級
            msg.setDelayTimeLevel(2);
            // 利用producer進(jìn)行發(fā)送，并同步等待發(fā)送結(jié)果
            SendResult sendResult = producer.send(msg, 10000);
            System.out.printf("%s%n", sendResult);
            // 一旦producer不再使用，關(guān)閉producer
            producer.shutdown();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

 @Test
    public void batchSend() {
        try {
            // 初始化一個(gè)producer并設(shè)置Producer group name
            DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
            // 設(shè)置NameServer地址
            producer.setNamesrvAddr(nameServer);
            // 啟動producer
            producer.start();
            List<Message> messages = new ArrayList<Message>();
            messages.add(new Message(topic, "Tag", "Order001", "Hello world 0".getBytes()));
            messages.add(new Message(topic, "Tag", "Order002", "Hello world 1".getBytes()));
            messages.add(new Message(topic, "Tag", "Order003", "Hello world 2".getBytes()));
            // 利用producer進(jìn)行發(fā)送，并同步等待發(fā)送結(jié)果
            SendResult sendResult = producer.send(messages, 10000);
            System.out.printf("%s%n", sendResult);
            // 一旦producer不再使用，關(guān)閉producer
            producer.shutdown();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

 @Test
    public void transactionSend() {
        try {
            // 事務(wù)消息的發(fā)送不再使用 DefaultMQProducer，而是使用 TransactionMQProducer 進(jìn)行發(fā)送
            TransactionMQProducer producer = new TransactionMQProducer(producerGroup);
            // 設(shè)置NameServer地址
            producer.setNamesrvAddr(nameServer);
            // 事務(wù)回查的線程池
            ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
                public Thread newThread(Runnable r) {
                    Thread thread = new Thread(r);
                    thread.setName("client-transaction-msg-check-thread");
                    return thread;
                }
            });
            producer.setExecutorService(executorService);
            producer.setTransactionListener(new TransactionListener() {
                //半事務(wù)消息發(fā)送成功后，執(zhí)行本地事務(wù)的方法
                public LocalTransactionState executeLocalTransaction(Message msg, Object o) {
                    System.out.printf("執(zhí)行本地事務(wù) %n");
                    /*
                    二次確認(rèn)
                    LocalTransactionState.COMMIT_MESSAGE：提交事務(wù)，允許消費(fèi)者消費(fèi)該消息
                    LocalTransactionState.ROLLBACK_MESSAGE：回滾事務(wù)，消息將被丟棄不允許消費(fèi)。
                    LocalTransactionState.UNKNOW：暫時(shí)無法判斷狀態(tài)，等待固定時(shí)間以后Broker端根據(jù)回查規(guī)則向生產(chǎn)者進(jìn)行消息回查。
                    */
                    return LocalTransactionState.UNKNOW;
                }
                // 二次確認(rèn)消息沒有收到，Broker端回查事務(wù)狀態(tài)的方法，默認(rèn)60s
                public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                    System.out.printf("二次確認(rèn)失敗，broker事務(wù)回查  %n");
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
            });
            producer.setSendMsgTimeout(10000);
            // 啟動producer
            producer.start();
            Message msg = new Message(topic, "tagA", "Hello RocketMQ".getBytes(RemotingHelper.DEFAULT_CHARSET));
            // 利用producer進(jìn)行發(fā)送事務(wù)消息，并同步等待發(fā)送結(jié)果
            SendResult sendResult = producer.sendMessageInTransaction(msg, null);
            System.out.printf("%s%n", sendResult);
            // 一旦producer不再使用，關(guān)閉producer
            System.in.read();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private final static String nameServer = "127.0.0.1:9876";
    private final static String consumerGroup = "my_group";
    private final static String topic = "topic-test";
    @Test
    public void consumerPush() throws MQClientException, IOException {
        // 初始化consumer，并設(shè)置consumer group name
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        // 設(shè)置NameServer地址
        consumer.setNamesrvAddr(nameServer);
        // 訂閱一個(gè)或多個(gè)topic，并指定tag過濾條件，這里指定*表示接收所有tag的消息
        consumer.subscribe(topic, "*");
        //設(shè)置采用廣播模式，廣播模式下，消費(fèi)組內(nèi)的每一個(gè)消費(fèi)者都會消費(fèi)全量消息。
        //consumer.setMessageModel(MessageModel.BROADCASTING);
        //注冊回調(diào)接口來處理從Broker中收到的消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                // 返回消息消費(fèi)狀態(tài)，ConsumeConcurrentlyStatus.CONSUME_SUCCESS為消費(fèi)成功
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 順序消費(fèi)
//        consumer.registerMessageListener(new MessageListenerOrderly() {
//            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
//                return ConsumeOrderlyStatus.SUCCESS;
//            }
//        });
        // 啟動Consumer
        consumer.start();
        System.out.printf("Consumer Started.%n");
        System.in.read();
    }

@Test
    public void consumerPull() {
        try {
            DefaultLitePullConsumer consumer = new DefaultLitePullConsumer(consumerGroup);
            consumer.setNamesrvAddr(nameServer);
            //關(guān)閉自動提交
            consumer.setAutoCommit(false);
            consumer.subscribe(topic, "*");
            consumer.setPullBatchSize(20);
            consumer.start();
            while (true) {
                List<MessageExt> messageExts = consumer.poll();
                System.out.printf("%s%n", messageExts);
                consumer.commitSync();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }