SpringBoot integra el código de muestra de RabbitMQ

Descripción general

1. RabbitMQ es un corredor de mensajes de código abierto y un servidor de cola que se utiliza para compartir datos entre aplicaciones completamente diferentes a través de un protocolo normal, o simplemente colegas de cola para procesar por un servidor distribuido.
2. Implementa de manera realista el protocolo AMQP y sigue el protocolo de código abierto de Licencia Pública de Mozilla, que admite múltiples idiomas y puede integrarse fácilmente con Spring.
3. Las colas de mensajes conectan aplicaciones utilizando mensajes que se enrutan entre aplicaciones a través de servidores proxy de mensajes como RabbitMQ.

Conceptos básicos

Corredor

Entidad del servidor de colas de mensajes utilizada para procesar datos

vhost

El host de mensaje virtual creado por el servidor RabbitMQ tiene su propio mecanismo de permiso. Se pueden abrir múltiples VHosts en un corredor de aislamiento de permisos para diferentes usuarios, y los VHosts también están completamente aislados.

Productor

Generar datos para la comunicación de mensajes

canal

Canal de mensajes, se pueden establecer múltiples canales en AMQP, cada canal representa una tarea de sesión.

intercambio

directo

Reenviar mensajes a la cola especificada por tecla de enrutamiento

previsión

previsión

Reenviar mensajes a todas las colas vinculadas es similar a una forma de transmisión.

tema

tema

Reenviar mensajes de acuerdo con las reglas. Esta regla es principalmente una coincidencia de patrones, y también parece más flexible.

cola

cola

1. La cola es un objeto interno de RabbitMQ, almacenando mensajes
2. Con los consumidores que aumentan dinámicamente, la cola asigna uniformemente los mensajes recibidos a múltiples consumidores de manera redonda.

vinculante

Representa la relación entre el interruptor y la cola. Al vincular, viene con una tecla de enlace de parámetro adicional para que coincida con la tecla de enrutamiento.

Consumidor

Escuche la cola de mensajes para leer los datos del mensaje

Tres modos de intercambio (fanot, directo, tema) Implementación bajo SpringBoot

Referencia de spring-boot-starter-amp en pom.xml

 <Spendency> <MoupRupid> org.springframework.boot </groupid> <artifactID> spring-boot-starter-amqp </artifactid> </pendency>

Agregar configuración de rabbitmq

 Primavera: Rabbitmq: Host: Localhost Port: 5672 Nombre de usuario: Invitado Contraseña: Invitado

directo

En modo directo, solo se requiere cola para definirla en general. Use el conmutador incorporado (defaultExchange) sin vincular el conmutador.

 @ConfigurationPublic Class RabbitP2PConfigure {public static final String queue_name = "P2P-Queue"; @Bean Public Queue Queue () {return new Queue (queue_name, true); }}

 @RunWith (SpringRunner.class) @SpringBoottest (classes = bootCoretestApplication.class)@slf4jpublic class rabBittest {@aUtowired private amqptemplate amqptEmplate; / *** Enviar*/ @Test public void sendlazy () lanza interruptedException {City City = New City (234556666l, "direct_name", "direct_code"); amqptemplate.convertandsend (rabbitlayAconfigure.queue_name, ciudad); } / *** recibir* / @test public void recibe () lanza interruptedException {object obj = amqptemplate.ReceiveAndConvert (rabBitLAyAconfigure.queue_name); Afirmar.notnull (obj, ""); log.debug (obj.ToString ()); }}

Escenarios aplicables: punto a punto

previsión

El modo de basura requiere unir múltiples colas al mismo interruptor

 @ConfigurationPublic Class RabbitFanoutConfigure {public static final String Exchange_Name = "Fanout-Exchange"; public static final String Fanout_a = "Fanout.a"; public static final String Fanout_b = "Fanout.b"; public static final String Fanout_c = "Fanout.c"; @Bean Public Queue amessage () {return new Queue (fanout_a); } @Bean public cole bMessage () {return new Queue (fanout_b); } @Bean public cole cmessage () {return new Queue (fanout_c); } @Bean public fanouTExchange fanouTExchange () {return New FanOutExchange (Exchange_Name); } @Bean public Binding BindingExChangea (Queue Amessage, FanOutExchange FanOutExchange) {return BindingBuilder.Bind (Amessage) .to (FanOutExchange); } @Bean public Binding BindingExchangeB (Queue BMessage, FanOutExchange FanOutExchange) {return BindingBuilder.Bind (BMessage) .to (FanOutExchange); } @Bean public Binding BindingExChangec (Queue CMessage, FanOutExchange FanOutExchange) {return bindingBuilder.bind (cMessage) .to (fanouTExchange); }}

Remitente

 @Slf4jpublic class remiter {@aUtowired private amqptemplate rabbittemplate; public void sendfanout (mensaje de objeto) {log.debug ("Comenzar el mensaje de fanotación <" + mensaje + ">"); rabbittemplate.convertandsend (rabbitfanoutconfigure.exchange_name, "", mensaje); }}

Podemos usar @rabbitListener para escuchar múltiples colas para consumir

 @Slf4j @rabbitListener (queues = {rabbitfanoutconfigure.fanout_a, rabbitfanoutconfigure.fanout_b, rabbitfanoutconfigure.fanout_c}) public class receptor {@rabbithandler public void void void void void whatesse }}

Escenarios aplicables
-Los juegos en línea de múltiples usuarios a gran escala (MMO) pueden usarlo para manejar eventos globales, como actualizaciones de clasificación
- Los sitios web de noticias deportivas pueden usarlo para distribuir actualizaciones de puntaje a clientes móviles en tiempo real
- El sistema de distribución lo utiliza para transmitir varios estados y actualizaciones de configuración
- Durante el chat grupal, se utiliza para distribuir mensajes a los usuarios que participan en el chat grupal.

tema

Este patrón es relativamente complejo. En pocas palabras, cada cola tiene su propio tema de preocupación. Todos los mensajes tienen un "título". Exchange reenviará los mensajes a colas cuyos temas se preocupan por los partidos difusos RouteKey.

Cuando sea vinculante, proporcione un tema que se preocupe por la cola, como "tema.# ("# "Significa 0 o varias palabras clave, y"*"significa una palabra clave).

 @ConfigurationPublic Class RabBittopicConfigure {public static final String Exchange_Name = "Topic-Exchange"; Public static final String Topic = "Topic"; public static final String topic_a = "topic.a"; public static final String topic_b = "topic.b"; @Bean Public Queue QueUetopic () {return New Queue (rabbittopicConfigure.topic); } @Bean public Queue Queuetopica () {return new Queue (rabbittopicConfigure.topic_a); } @Bean public Queue QueUetopicB () {return new Queue (rabbittópicoConfigure.topic_b); } @Bean public topicExchange Exchange () {topiceXchange topiceXchange = new topiceXchange (Exchange_Name); topicExchange.setDelayed (verdadero); devolver el nuevo topiceXchange (Exchange_Name); } @Bean public Binding BindingExChangeTopic (Queue Queuetópico, TopiceXchange Exchange) {return bindingBuilder.bind (queuetópico) .to (intercambio) .with (rabbittópicoConfigure.topic); } @Bean public Binding BindingExChangeTopics (Queue Queuetopica, TopicExchange Exchange) {return bindingBuilder.bind (queuetopica) .to (intercambio) .with ("tema.#"); }}

Al mismo tiempo, escucha tres colas

 @Slf4j @rabbitListener (queues = {rabBittopicConfigure.topic, rabBittopicConfigure.topic_a, rabBittopicConfigure.topic_b}) receptor de clase pública {@rabbithandler public void recibeMessage (mensaje de cadena) }}

A través de las pruebas podemos encontrar

 @Runwith (SpringRunner.class) @SpringBoottest (classes = bootCoretestApplication.class) clase pública Rabbittest {@aUtowired private amqptemplate rabbittemplate; @Test public void sendall () {rabbittemplate.convertandsend (rabbittopicConfigure.exchange_name, "topic.test", "enviar todo"); } @Test public void sendTopic () {rabbittemplate.convertandsend (rabbittopicConfigure.exchange_name, rabBittopicConfigure.topic, "Enviar tema"); } @Test public void sendTopica () {rabbittemplate.convertandsend (rabbittopicConfigure.exchange_name, rabBittopicConfigure.topic_a, "Enviar topica"); }}

Escenarios aplicables
- Distribuir datos sobre ubicaciones geográficas específicas, como el punto de venta
- Tareas detrás del escenario completadas por varios trabajadores, cada trabajador responsable de manejar ciertas tareas específicas
- Actualizaciones del precio de las acciones (y otros tipos de actualizaciones de datos financieros)
- Actualizaciones de noticias que involucran categorías o etiquetas (por ejemplo, para deportes o equipos específicos)
- Coordinación de diferentes tipos de servicios en la nube
- Arquitectura distribuida/paquete de software basado en el sistema, donde cada constructor solo puede manejar una arquitectura o sistema específico.

Retraso

Consumo tardío:

1. Si el usuario genera un pedido, tomará un período de tiempo verificar el estado de pago del pedido. Si el pedido aún no se paga, el pedido debe cerrarse a tiempo.
2. Después de que el registro del usuario sea exitoso, tomará un período de tiempo, como una semana para verificar el uso del usuario. Si se encuentra que la actividad del usuario es baja, envíe un correo electrónico o un mensaje de texto para recordarle al usuario que la use.

Reintento retrasado:

1. Si el consumidor no consume mensajes de la cola, pero quiere volver a intentarlo automáticamente después de retrasarse por un período de tiempo.
2. Si no usamos la cola de retraso, solo podemos hacerlo a través de un escáner de votación. Esta solución no es elegante ni conveniente para hacer un servicio unificado para los desarrolladores. Pero con una cola de retraso, podemos hacerlo fácilmente.

Establezca la propiedad de retraso de cambio en verdadero

 @ConfigurationPublic Class RabbitLAyAyConfigure {public static final String queue_name = "LAZAY-QUEUE-T"; public static final String Exchange_Name = "Lazy-Exchange-T"; @Bean Public Queue Queue () {return new Queue (queue_name, true); } @Bean public DirectExchange DefaultExchange () {DirectExchange DirectExchange = new DirectExchange (Exchange_Name, True, False); directExchange.setDelayed (verdadero); return DirectExchange; } @Bean public Binding Binding () {return bindingBuilder.bind (queue ()). A (defaultExchange ()). Con (queue_name); }}

Establezca el tiempo de retraso al enviar

 @Slf4jpublic class remiter {@aUtowired private amqptemplate rabbittemplate; public void sendlazy (msg de objetos) {log.debug ("Comenzar envía un mensaje perezoso <" + msg + ">"); rabbittemplate.convertAndSend (rabbitlayAconfigure.exchange_name, rabbitlazyConfigure.queue_name, msg, mensaje -> {message.getMessageProperties (). Setheader ("X -DELAY", 10000); return Message;}); }}

Finalizar

Consulte los documentos oficiales directamente para obtener varios casos de uso.

Lo anterior es todo el contenido de este artículo. Espero que sea útil para el aprendizaje de todos y espero que todos apoyen más a Wulin.com.