package tttt.mq;import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.junit.Test;public class MqProductTest {@Testpublic void test1() {DefaultMQProducer producer = new DefaultMQProducer("xiaof_test");producer.setNamesrvAddr("192.168.0.128:9876");try {producer.start();for (int i = 0; i < 2; i++)try {{Message msg = new Message("Topic1", "TagA", "OrderID188","Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
//                      SendResult sendResult = producer.send(msg);
//                      System.out.printf("%s%n", sendResult);producer.sendOneway(msg);}} catch (Exception e) {e.printStackTrace();}} catch (MQClientException e) {e.printStackTrace();} finally {producer.shutdown();}}}

package tttt.mq;import java.util.List;
import java.util.concurrent.atomic.AtomicLong;import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Test;public class MqConsumeTest {@Testpublic void test1() {DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("xiaof_test");consumer.setNamesrvAddr("192.168.0.128:9876");consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);try {consumer.subscribe("Topic1", "TagA");consumer.registerMessageListener(new MessageListenerOrderly() {AtomicLong consumeTimes = new AtomicLong(0);@Overridepublic ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {//这个要是false，服务器就会不断重复发送消息context.setAutoCommit(true);MessageExt msg = msgs.get(0);String data = new String(msg.getBody());System.out.printf("%s 消费信息线程与数据: %s %n", Thread.currentThread().getName(), data);this.consumeTimes.incrementAndGet();if ((this.consumeTimes.get() % 2) == 0) {return ConsumeOrderlyStatus.SUCCESS;} else if ((this.consumeTimes.get() % 3) == 0) {return ConsumeOrderlyStatus.ROLLBACK;} else if ((this.consumeTimes.get() % 4) == 0) {return ConsumeOrderlyStatus.COMMIT;} else if ((this.consumeTimes.get() % 5) == 0) {context.setSuspendCurrentQueueTimeMillis(3000);return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;}return ConsumeOrderlyStatus.SUCCESS;}});consumer.start();System.out.printf("Consumer Started.%n");} catch (MQClientException e) {e.printStackTrace();}}
}