Shulou(Shulou.com)06/02 Report--

Preface

Message queue is very commonly used in today's systems with large amount of data and high concurrency. This article will analyze and compare some of the most commonly used message queue frameworks: ActiveMQ, RabbitMQ and Kafka.

The structure and implementation principle of RabbitMQ under Spring framework are introduced in detail, from the transaction and callback function (ConfirmCallback / ReturnCallback) on the Producer side to the MessageListenerContainer information receiving container on the Consumer side. Through the introduction of RabbitTemplate, SimpleMessageListenerContainer, DirectMessageListenerContainer and other common types, this paper deeply analyzes the principle and matters needing attention in each transmission link of message processing.

At the same time, the dead letter queue and persistence operation are introduced one by one with examples.

Catalogue

I. the relationship between RabbitMQ and AMQP

Second, the realization principle of RabbitMQ

Third, RabbitMQ application example

IV. Message sending and monitoring on the Producer side

Management and control of Consumer message reception

VI. Dead letter queue

7. Persistence operation

I. the relationship between RabbitMQ and AMQP

AMQP (Advanced Message Queue Protocol Advanced message queuing Protocol) is a message queuing protocol that supports qualified clients to communicate with message broker middleware (message middleware broker). RabbitMQ is the implementer of AMQP protocol, which is mainly used to store, send and receive information in distributed systems. The server side of RabbitMQ is written in Erlang language, and the client supports a variety of development languages: Python, .NET, Java, Ruby, C, PHP, ActionScript, XMPP, STOMP and so on.

1.1 comparison of ActiveMQ, RabbitMQ and Kafka

Now there are many commonly used message queuing frameworks in the market, such as ActiveMQ, RabbitMQ, Kafka and so on. Compared with other frameworks, RabbitMQ has good performance in ease of use, scalability, high availability, multi-protocol, support for multilingual clients and so on.

1.2.1 characteristics of AcitveMQ

ActiveMQ is a message model developed by Apache in Java language. It perfectly supports JMS (Java Message Service) message service. The client supports Java, C, C++, C #, Ruby, Perl, Python, PHP and other open and host languages, and supports OpenWire, Stomp, REST, XMPP, AMQP and other protocols. ActiveMQ adopts asynchronous message passing mode, which ensures the effective communication of multi-host cluster, client-server, point-to-point and other modes in design. From the beginning, it has been developed in accordance with JMS 1.1 and J2EE 1.4 specifications, implementing message persistence, XA, transaction support and other functions. After years of upgrading and improvement, it has become the mainstream messaging solution in Java application development. However, compared with RabbitMQ and Kafka, its main disadvantages are relatively large resource consumption, low throughput and weak system supporting capacity in the case of high concurrency. If the system is developed using Java, its concurrency is within a controllable range, or the system needs to support a variety of different protocols, ActiveMQ can be used to build a message queue service more easily.

1.2.2 characteristics of Kafka

Kafka is born as a message queue for distributed system development. It has the characteristics of high performance, disaster tolerance, dynamic expansion and so on. The inherent characteristic of Kafka is that it backs up the data of each Partition to a different server and cooperates with ZooKeeper. When a Broker failure fails, the ZooKeeper service will notify the producer and consumer to recover the data from the backup server. In terms of performance, Kafka also greatly surpasses the traditional ActiveMQ and RabbitMQ. Because the Kafka cluster can support dynamic expansion, new servers can be dynamically added to the cluster when the load reaches its peak without restarting the service. However, because Kafka is a distributed system, it can only achieve message ordering in the same partition, not global message ordering. And its internal monitoring mechanism is not perfect, so it needs to install plug-ins and rely on ZooKeeper for metadata management. If the system is a distributed management mechanism, the amount of data is large and the concurrency is difficult to predict, it is recommended to use Kafka queue.

1.2.3 RabbitMQ comparison

Because ActiveMQ relies too much on the specification of JMS and limits its development, RabbitMQ will obviously outperform ActiveMQ in terms of performance and throughput.

Due to the long time on the market, RabbitMq will be more mature than Kafka in terms of availability, stability and reliability, and RabbitMq is developed using Erlang, so it is born with the characteristics of high concurrency and high availability. Kafka is a distributed system, its performance, throughput, TPS will be better than RabbitMq.

Second, the realization principle of RabbitMQ

2.1 the relationship among producers (Producer), consumers (Consumer) and service centers (Broker)

First of all, a brief introduction of the operation principle of RabbitMQ, in the use of RabbitMQ, the system will first install and start Broker Server, that is, the service center of RabbitMQ. Both producers (Producer) and consumers (Consumer) use the TCP/IP protocol (default) to connect to BrokerServer through connection pooling (Connection). Then Producer will send the binding information of Exchange / Queue to Broker Server,Broker Server to select the corresponding Queue according to the type logic of Exchange, and finally send the information to the corresponding Consumer associated with Queue.

2.2 Concepts of switch (Exchange), queue (Queue), channel (Channel) and binding (Binding)

2.2.1 switch Exchange

After Producer establishes a connection, it does not directly deliver the message to the queue Queue, but sends the message to the switch Exchange, and the Exchange sends the message to one or more corresponding queues according to different logic. Currently, Exchange provides four different common types: Fanout, Direct, Topic, and Header.

Fanout Typ

This type is the most common switch, which forwards messages to all queues bound to it. For example, there are N queues bound to the Fanout switch, and when a message is generated, Exchange sends N copies of the message to each queue, similar to the broadcast mechanism.

Direct Typ

This type of Exchange sends messages to queues with exactly the same Routing_Key. Multiple Cousumer can be bound with the same keyword, similar to the one-to-many relationship of a database. For example, Producer uses Exchange of type Direct to push Routing_Key as a queue for direct.key1, and the system specifies multiple Cousumer binding direct.key1. In this way, the message is distributed to several different Cousumer.

Topic Typ

This type is the most flexible way to configure, and it can use fuzzy matching to bind to different queues that contain the keyword based on the Routing_Key. For example, Producer uses Exchange of Topic type to push different queues with Routing_Key set to topic.guangdong.guangzhou and topic.guangdong.shenzhen respectively. Cousumer only needs to set Routing_Key to topic.guangdong.# to receive and process all messages.

Headers Typ

This type of switch is slightly different from the one described earlier in that it is no longer routed based on the keyword Routing_Key, but on multiple attributes that are more easily represented as headers of the message than routing keywords. That is, the attribute used for routing is taken from the message Header property, and when the value of the message Header is the same as that specified when the queue is bound, the message is routed to the corresponding queue.

2.2.2 Queue queue

Queue queue is the carrier of messages, each message will be put into the Queue, it contains name,durable,arguments and other attributes, name is used to define its name, when the durable (persistence) is true, the queue will be persisted to the hard disk. When it is false, once the Broker Server is restarted, the corresponding queue will disappear, which will be described in detail later.

2.2.3 Channel Channel

When Broker Server uses Connection to connect to Producer / Cousumer, it uses Channel. Multiple Channel can be established on a Connection, and each Channel has a session task, which can be understood as a logical connection. Mainly used for management-related parameter definition, sending messages, getting messages, transaction processing and so on.

2.2.4 Binding binding

Binding is mainly used to bind the correspondence between the switch Exchange and the queue Queue and to record the Routing-Key of the route. The Binding information is saved to the system and used as the basis for the distribution of Broker Server information.

Third, RabbitMQ application example

3.1 Rabbit Common Class description

3.1.1 RabbitTemplate class

The Spring framework has encapsulated RabbitTemplate to simplify the management of RabbitMQ binding, queue sending and receiving.

Method description: void setExchange (String exchange) sets the bound exchange name String getExchange () gets the bound exchange name void setRoutingKey (String routingKey) sets the bound routingKeyString getRoutingKey () gets the bound routingKeyvoid send (String exchange, String routingKey, Message message,CorrelationData data) sends information to Broken Server,CorrelationData in Message mode as an identifier can be empty void convertAndSend (String exchange, String routingKey, Object object, CorrelationData data) sends information to Broken Server in custom object mode, the system will automatically convert object to Message CorrelationData can be empty Message receive (String queueName, long timeoutMillis) send Message information Object receiveAndConvert (String queueName) according to queueuName receive queue Long timeoutMillis) based on queueuName receive queue object information void setReceiveTimeout (long receiveTimeout) set receive expiration time void setReplyTimeout (long replyTimeout) set retransmission time void setMandatory (boolean mandatory) enable mandatory delegation mode (more on below) void setConfirmCallback (confirmCallback) bind message confirm callback method (more on below) void setReturnCallback (returnCallback) bind message exit callback method (more on below)

3.2 preliminary study of RabbitMQ

After downloading and successfully installing RabbitMQ on the official website, open the default path http://localhost:15672/#/ to see the management interface of the RabbitMQ service center

3.2.1 Producer side development

First add the dependency of RabbitMQ in pom, and add information such as RabbitMQ account password and so on in application.yml. For this example, we try to use the Direct switch to send queues to different Consumer.

1 * * pom * 2 3. 45 org.springframework.boot 6 spring-boot-starter-amqp 7 2.0.5.RELEASE 8 9 10 11 * application.yml * 12 spring:13 application:14 name: rabbitMqProducer15 rabbitmq:16 host: localhost 17 port: 567218 username: admin19 password: 1234567820 virtual-host: / LeslieHost

First, use CachingConnectionFactory to establish a link and bind the relationship between Exchange, Queue, and RoutingKey through BindingBuilder.

Then send the information to Broken Server through the void convertAndSend (String exchange, String routingKey, Object object, CorrelationData data) method

1 @ Configuration 2 public class ConnectionConfig {3 @ Value ("${spring.rabbitmq.host}") 4 public String host; 5 6 @ Value ("${spring.rabbitmq.port}") 7 public int port; 8 9 @ Value ("${spring.rabbitmq.username}") 10 public String username;11 12 @ Value ("${spring.rabbitmq.password}") 13 public String password 14 15 @ Value ("${spring.rabbitmq.virtual-host}") 16 public String virtualHost;17 18 @ Bean19 public ConnectionFactory getConnectionFactory () {20 CachingConnectionFactory factory=new CachingConnectionFactory (); 21 factory.setHost (host); 22 factory.setPort (port); 23 factory.setUsername (username); 24 factory.setPassword (password); 25 factory.setVirtualHost (virtualHost); 26 return factory 27} 28} 29 30 @ Configuration31 public class BindingConfig {32 public final static String first= "direct.first"; 33 public final static String second= "direct.second"; 34 public final static String Exchange_NAME= "directExchange"; 35 public final static String RoutingKey1= "directKey1"; 36 public final static String RoutingKey2= "directKey2"; 37 38 @ Bean39 public Queue queueFirst () {40 return new Queue (first) 41} 42 43 @ Bean44 public Queue queueSecond () {45 return new Queue (second); 46} 47 48 @ Bean49 public DirectExchange directExchange () {50 return new DirectExchange (Exchange_NAME,true,true); 51} 52 53 / / bind Direct to queueFirst54 @ Bean55 public Binding bindingExchangeFirst (Queue queueFirst, DirectExchange directExchange) {56 return BindingBuilder.bind (queueFirst) .to (directExchange) .with (RoutingKey1) using BindingBuilder 57} 58 59 / / bind Direct to queueSecond60 @ Bean61 public Binding bindingExchangeSecond (Queue queueSecond, DirectExchange directExchange) {62 return BindingBuilder.bind (queueSecond) .to (directExchange) .with (RoutingKey2) using BindingBuilder; 63} 64} 65 66 @ Controller67 @ RequestMapping ("/ producer") 68 public class ProducerController {69 @ Autowired70 private RabbitTemplate template;71 72 @ RequestMapping ("/ send") 73 public void send () {74 for (int nasty 0) N

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.