生產者：
int orderId = 1000;//訂單號

DefaultMQProducer orderProducer = new DefaultMQProducer("orderProducerGroup");
orderProducer.setNamesrvAddr("ip1:port;ip2:port");
orderProducer.start();
for(int i = 0; i < 10; i ++){
    Message msg = new Message("topic", "tags", "keys" + i, "body".getBytes(RemotingHelper.DEFAULT_CHARSET));
    orderProducer.send(msg, new MessageQueueSelector(){
    @Override
    public MessageQueue select(List mqs, Message msg, Object arg){
        Integer id = (Integer)arg;
        int index = id%mqs.size();
        return mqs.get(index);
    }
}, orderId);
}
orderProducer.shutdown();

消費者：

DefaultMQPushConsumer pushConsumer = new DefaultMQPushConsumer("pushConsumerGroup");
pushConsumer.setNamesrvAddr("ip1:port;ip2:port");
pushConsumer.subscribe("topic", "*");//訂閱主題
pushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
pushCosumer.setMessageModel(MessageModel.CLUSTERING);//叢集消費、預設也是“叢集消費”
pushConsumer.registerMessageListener(new MessageListenerOrderly(){
    @Override
    public ConsumeOrderStatus consumeMessage(List msgs, ConsumeOrderlyContext context){
        for(int i = 0; i < msgs.size(); i ++){
            Message msg = msgs.get(i);
            System.err.println("topic:" + msg.getTopic() + ",keys:" + msg.getKeys());
        }
        return ConsumeOrderStatus.SUCCESS;
    }
});
pushConsumer.start();




 

DefaultMQProducer batchProducer = new DefaultMQProducer("bathProducerGroup");
batchProducer.setNamesrvAddr("ip1:port;ip2:port");
batchProducer.start();
List msgs = new ArrayList();
msgs.add(msg1);
msgs.add(msg2);
..........
batchProducer.send(msgs);
batchProducer.shutdown();
 

生產者：

DefaultMQProducer delayProducer = new DefaultMQProducer("delayProducerGroup");
delayProducer.setNamesrvAddr("ip1:port;ip2:port");
delayProducer.start();

Message delayMsg = new Message("topic", "tags" , "keys", "binary content");
delayMsg.setDelayTimeLevel(3);//延遲10s被消費
delayProducer.send(delayMsg);

delayProducer.shutdown();



消費者：

DefaultMQConsumer consume = new DefaultMQConsumer("consumeGroup");
consume.setNamesrvAddr("ip1:port;ip2:port");
consume.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consume.registerMessageListener(new MessageListenerConcurrently(){
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List msgs, ConsumeConcurrentlyContext context){
        for(int i = 0; i < msgs.size(); i ++){
            Message msg = msgs.get(i);
            System.out.println("topic:" + msg.getTopic() + "keys:" + msg.getKeys());
        }
        return ConsumeConcurrentlyContext.CONSUME_SUCCESS;
    }
});