Rockermq message sending and consumption mode

Catalog

Message sending mode

brief introduction

Code

Differences among three transmission modes

Message consumption mode

brief introduction

Cluster consumption mode

Introduce

Code

step  

Broadcast consumption mode

Introduce

demonstration

---

Message sending mode

brief introduction

RocketMQ There are three sending modes , They are synchronous transmission 、 Send asynchronously 、 Send one way , Different patterns are applicable to different business scenarios

Code

public static void main(String[] args) {
	DefaultMQProducer defaultProducer = getDefaultProducer();
	Producer producer = new Producer();
	producer.Sync(defaultProducer);
	producer.Async(defaultProducer);
	producer.oneway(defaultProducer);
	defaultProducer.shutdown();
}
public static DefaultMQProducer getDefaultProducer() {
	try {
		DefaultMQProducer producer = new DefaultMQProducer("producerGroup");
		producer.setNamesrvAddr("localhost:9876");
		producer.start();
		return producer;
	} catch (Exception e) {
		e.printStackTrace();
		return null;
	}
}
/**
 *  The synchronous 
 * @param producer
 */
public void Sync(DefaultMQProducer producer) {
	try {
		//  After the synchronization message fails to be sent , Resend several times 
		producer.setRetryTimesWhenSendFailed(0);
		Message msg = new Message("topic", " The synchronous ".getBytes());
		SendResult res = producer.send(msg);
		System.out.println("res" + res);
	} catch (Exception e) {
		e.printStackTrace();
	}
}
/**
 *  Send asynchronously 
 * @param producer
 */
public void Async(DefaultMQProducer producer) {
	try {
		//  After asynchronous message sending fails , Resend several times 
		producer.setRetryTimesWhenSendAsyncFailed(0);
		Message msg = new Message("topic", " Asynchronous messaging ".getBytes());
		producer.send(msg, new SendCallback() {
			@Override
			public void onSuccess(SendResult sendResult) {
				System.out.println("sendResult:" + sendResult);
			}
			@Override
			public void onException(Throwable throwable) {
				System.out.println("throwable:" + throwable);
			}
		});
	} catch (Exception e) {
		e.printStackTrace();
	}
}
/**
 *  Send one way 
 * @param producer
 */
public void oneway(DefaultMQProducer producer) {
	try {
		Message msg = new Message("topic", " Send one way ".getBytes());
		producer.sendOneway(msg);
	}catch (Exception e) {
		e.printStackTrace();
	}
}

Differences among three transmission modes

The synchronous ： Message sent to master broker And sync to slave broker after , Will respond to the client , Slow efficiency , But the risk of losing data is small

Send asynchronously ： Message sent to master broker And then respond to the client , No need to wait for a successful sync to slave broker, Efficient , The risk is high . for example master broker After processing the message , After responding to the client , Not synchronized to slave broker

Send one way ： Producers only need to produce messages , There is no need to broker Return results , The most efficient , The highest risk , Applicable to scenarios where message loss is allowed

Message consumption mode

brief introduction

RocketMQ There are two consumption patterns , Cluster （CLUSTERING） And broadcast （BROADCASTING）, The default is cluster mode . Consumption patterns are defined by consumers .

Cluster consumption mode

Introduce

Cluster consumption mode means that messages are consumed by only one consumer in the cluster , If there are multiple clusters , Each cluster will consume only once , When a message is re delivered, it cannot be guaranteed that it will be routed to the same machine , Message status by broker maintain

Code

public static void main(String[] args) throws Exception {
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumerGroup");
    consumer.setNamesrvAddr("localhost:9876");
    consumer.subscribe("topic", "*");
    consumer.setMessageModel(MessageModel.CLUSTERING);
    consumer.registerMessageListener((MessageListenerConcurrently) (msgList, context) -> {
        for (MessageExt msg : msgList) {
            System.out.println(new String(msg.getBody()));;
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    });
    consumer.start();
    System.out.println("ConsumerA start...");
}

step  

Start the two consumers with the above code A、B

Use the producer code to send a message or create... In the monitoring platform topic And send a message

The steps for the monitoring platform to send messages are as follows

As a result, only one consumer will receive the message  

Broadcast consumption mode

Introduce

Broadcast messages are messages that are pushed to all currently registered consumers in the cluster , If there is no consumer online at present, it will wait until a consumer pulls a message , But if consumption fails, it will not be reinvested

demonstration

Put the above code to MessageModel Switch to BROADCASTING that will do

consumer.setMessageModel(MessageModel.BROADCASTING);

The other steps are the same