RocketMQ-消费MQ消息简单示例

public static void main(String[] args) throws InterruptedException, MQClientException {
	DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test-consumer");
	consumer.setNamesrvAddr("1.1.1.1:9876");
	consumer.subscribe("test-topic", "*");
	consumer.registerMessageListener(new MessageListenerConcurrently() {
		@Override
		public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,  ConsumeConcurrentlyContext context) {
			System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
			try {
				for (MessageExt messageExt : msgs){
					String messageBody = new String(messageExt.getBody(), "utf-8");
					System.out.println(messageBody);
				}
			} catch (UnsupportedEncodingException e) {
				e.printStackTrace();
			}
			return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
		}
	});
	consumer.start();
	System.out.printf("Consumer Started.");
}

执行结果：

Consumer Started.
ConsumeMessageThread_1 Receive New Messages: [MessageExt [queueId=2, storeSize=351, ... ] 
ConsumeMessageThread_2 Receive New Messages: [MessageExt [queueId=0, storeSize=351, ...] 
{"ticketFailure":27,"total":157,"airTicketTotal":115...}