Springboot интегрирует код выборки RabbitMQ

Обзор

1. RabbitMQ - это брокер с открытым исходным кодом и сервер Queue, который используется для обмена данными между совершенно разными приложениями с помощью обычного протокола, или просто задания очереди для обработки распределенным сервером.
2. Он реалистично реализует протокол AMQP и следует за протоколом с открытым исходным кодом Mozilla с открытым исходным кодом, который поддерживает несколько языков и может быть легко интегрирован с пружиной.
3. Ожидания сообщений подключают приложения с использованием сообщений, которые маршрутизируются между приложениями через прокси -серверы сообщений, такие как Rabbitmq.

Основные понятия

Маклер

Предприятие сервера очередей сообщений используется для обработки данных

Vhost

Виртуальный хост сообщений, созданный сервером RabbitMQ, имеет свой собственный механизм разрешения. Несколько Vhosts могут быть открыты в брокере для изоляции разрешений для разных пользователей, а Vhosts также полностью изолированы.

Продукт

Создать данные для сообщения сообщений

канал

Канал сообщений, несколько каналов могут быть установлены в AMQP, каждый канал представляет задачу сеанса.

обмен

прямой

Переместить сообщения в очередь, указанную путем маршрутизации

фанату

фанату

Перестановление сообщений во все связанные очереди аналогична способу вещания.

тема

тема

Переслать сообщения в соответствии с правилами. Это правило в основном соответствует моделям, и оно также кажется более гибким.

очередь

очередь

1. Очередь является внутренним объектом RabbitMQ, храня сообщения
2. С динамическим увеличением потребителей очередь равномерно распределяет полученные сообщения нескольким потребителям в раунде.

связывание

Он представляет связь между переключением и очередью. При привязке он поставляется с дополнительным параметрами для привязки к ключе, чтобы соответствовать ключе маршрутизации.

Потребитель

Слушайте очередь сообщений, чтобы прочитать данные сообщения

Три режима обмена (Fanout, Direct, тема) реализация под Springboot

Справочный Spring-Boot-Starter-AMQP в pom.xml

 <Depective> <groupid> org.springframework.boot </GroupId> <artifactid> Spring-boot-starter-amqp </artifactid> </dependency>

Добавить конфигурацию RabbitMQ

 Весна: Rabbitmq: Хост: Localhost Port: 5672 Имя пользователя: пароль гостя: гость

прямой

В прямом режиме требуется только очередь для определения ее в целом. Используйте встроенный переключатель (Defaultexchange) без привязки переключателя.

 @Configurationpublic class rabbitp2pconfigure {public static final String queue_name = "p2p-queue"; @Bean public queue queue () {вернуть новую очередь (queue_name, true); }}

 @Runwith (springrunner.class) @springboottest (classes = bootcoretestapplication.class)@slf4jpublic class rabbittest {@autowired private amqptemplate amqptemplate; / *** Отправить*/ @test public void sendlazy () бросает прерванную экспрессию {City City = New City (2345566666L, "Direct_Name", "Direct_Code"); amqptemplate.convertandsend (rabbitlazyconfigure.queue_name, City); } / *** Получить* / @test public void checept () бросает прерывание ExtruptException {Object obj = amqptemplate.receiveAndconvert (rabbitlazyconfigure.queue_name); Assert.notnull (obj, ""); log.debug (obj.tostring ()); }}

Применимые сценарии: точка-точка

фанату

Режим вентилятора требует привязки нескольких очередей с одним и тем же переключателем

 @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 () {вернуть новую очередь (fanout_a); } @Bean public queue bmessage () {вернуть новую очередь (fanout_b); } @Bean public queue cmessage () {вернуть новую очередь (fanout_c); } @Bean public fanoutexchange fanoutexchange () {return new fanoutexchange (Exchange_name); } @Bean public bindingexchangea (queue amessage, fanoutexchange fanoutexchange) {return bindingbuilder.bind (amessage) .to (fanoutexchange); } @Bean public bindingexchangeb (queue bmessage, fanoutexchange fanoutexchange) {return bindingbuilder.bind (bmessage) .to (fanoutexchange); } @Bean public bindingexchangec (queue cmessage, fanoutexchange fanoutexchange) {return bindingbuilder.bind (cmessage) .to (fanoutexchange); }}

Отправитель

 @Slf4jpublic class sender {@autowired private amqptemplate rabbittemplate; public void sendfanout (Object Message) {log.debug ("Начать отправление отправки сообщения fanout <" + message + ">"); rabbittemplate.convertandsend (rabbitfanoutconfigure.exchange_name, "", сообщение); }}

Мы можем использовать @rabbitlistener, чтобы прослушать несколько очередей для потребления

 @Slf4j @RabbitListener (queues = {rabbitfanoutConfigure.fanout_a, RabbitfanoutConfigure.fanout_b, RabbitfanoutConfigure.fanout_c}) Public Class Receiver {@rabbithandler public Void PectiveMessage (строковое сообщение) {log.de ("Познакомится <) }}

Применимые сценарии
-крупномасштабные многопользовательские онлайн-игры (MMO) могут использовать их для обработки глобальных событий, таких как обновления ранжирования
- Спортивные новостные веб -сайты могут использовать его для распространения обновлений с партитурой для мобильных клиентов в режиме реального времени
- Система распределения использует его для трансляции различных состояний и обновлений конфигурации
- Во время группового чата он используется для распространения сообщений пользователям, участвующим в групповом чате.

тема

Этот шаблон относительно сложный. Проще говоря, каждая очередь имеет свою собственную тему, вызывающую беспокойство. Все сообщения имеют «заголовок». Exchange переназначит сообщения в очередь, темы которых касаются нечетких матчей Routekey.

При обязательном отношении предоставьте тему, о которой касается очередь, такой как «Тема.# ("# "Означает 0 или несколько ключевых слов, а"*"" означает ключевое слово.)

 @Configurationpublic class rabbittopicconfigure {public static final String Exchange_Name = "topic-exchange"; Общественная статическая конечная строка topic = "topic"; Общественная статическая конечная строка TOMA_A = "TOMA.A"; Общественная статическая конечная строка TOMA_B = "TOMA.B"; @Bean public queue queuetopic () {вернуть новую очередь (rabbittopicconfigure.topic); } @Bean public queue queuetopica () {вернуть новую очередь (rabbittopicconfigure.topic_a); } @Bean public queue queuetopicb () {return new queue (rabbittopicconfigure.topic_b); } @Bean public tapicexchange Exchange () {tapicexchange tapicexchange = new tapicexchange (Exchange_name); tapicexchange.setDelayed (true); вернуть новый TapicexChange (Exchange_Name); } @Bean public bindingExchangeTopic (queue queuetopic, tapicexchange Exchange) {return BindingBuilder.bind (queuetopic) .to (Exchange) .with (rabbittopicConfigure.topic); } @Bean public bindingexchangeTopics (queue queuetopica, tapicexchange Exchange) {return BindingBuilder.bind (Queuetopica) .to (Exchange) .with ("Theme.#"); }}

В то же время послушайте три очереди

 @Slf4j @RabbitListener (queues = {rabbittopicConfigure.topic, rabbittopicConfigure.topic_a, RabbittopicConfigure.topic_b}) Public Class Receiver {@rabbithandler public veodegessage (строковое сообщение) {log.debug ("Полученное сообщение +" + " +" + " +" + " +" + " +" + " +" + " +" + " +" + " +" + " +" + " +" + " +" + " +" + " +" + " +" + " +" + " +" + " +" + " +" + " +" + " +" + " +" + " +" }}

Через тестирование мы можем найти

 @Runwith (springrunner.class) @springboottest (classes = bootcoretestapplication.class) public class rabbittest {@autowired private amqptemplate rabbittemplate; @Test public void sendall () {rabbittemplate.convertandsend (rabbittopicconfigure.exchange_name, "topic.test", "Отправить все"); } @Test public void sendtopic () {rabbittemplate.convertandsend (rabbittopicconfigure.exchange_name, rabbittopicconfigure.topic, "Отправить тему"); } @Test public void sendtopica () {rabbittemplate.convertandsend (rabbittopicconfigure.exchange_name, rabbittopicconfigure.topic_a, "Отправить топику"); }}

Применимые сценарии
- Распределить данные о конкретных географических местоположениях, таких как точка продажи
- Закуливые задачи, выполненные несколькими работниками, каждый работник, отвечающий за выполнение определенных конкретных задач
- Обновления цен на акции (и другие виды обновлений финансовых данных)
- Обновления новостей с участием категорий или тегов (например, для конкретных видов спорта или команд)
- Координация различных типов сервисов в облаке
- Распределенная архитектура/системный программный пакет, где каждый строитель может обрабатывать только одну конкретную архитектуру или систему.

Очередь задержки

Отсроченное потребление:

1. Если пользователь генерирует заказ, потребуется период времени, чтобы проверить статус оплаты заказа. Если заказ еще не оплачен, заказ должен быть закрыт вовремя.
2. После успеха пользователя потребуется период времени, например, неделя, чтобы проверить использование пользователя. Если обнаружено, что деятельность пользователя низкая, отправьте электронное письмо или текстовое сообщение, чтобы напомнить пользователю использовать его.

Задержка повторения:

1. Если потребитель не может употреблять сообщения из очереди, но хочет автоматически повторить попытку после задержки в течение определенного периода времени.
2. Если мы не используем очередь задержки, мы можем сделать это только с помощью опроса. Это решение не является ни элегантным, ни удобным, чтобы сделать унифицированную услугу для разработчиков. Но с очередью задержки мы можем сделать это легко.

Установите свойство задержки Switch на True

 @Configurationpublic class rabbitlazyconfigure {public static final String queue_name = "Lazy-queue-t"; public Static Final String Exchange_Name = "Lazy-Exchange-T"; @Bean public queue queue () {вернуть новую очередь (queue_name, true); } @Bean public directExchange defaultexchange () {DirectExchange DirectExchange = new DirectExchange (Exchange_Name, true, false); DirectExchange.SetDelayed (TRUE); вернуть DirectExchange; } @Bean public spinting sinting () {return bindingbuilder.bind (queue ()). }}

Установите время задержки при отправке

 @Slf4jpublic class sender {@autowired private amqptemplate rabbittemplate; public void sendlazy (Object msg) {log.debug («Начните отправить Lazy Message <» + msg + ">"); rabbittemplate.convertandsend (rabbitlazyconfigure.exchange_name, rabbitlazyconfigure.queue_name, msg, message -> {message.getmessageproperties (). Setheader ("x -delay", 10000); return message;}); }}

Заканчивать

Пожалуйста, проверьте официальные документы напрямую на наличие различных вариантов использования

Выше всего содержание этой статьи. Я надеюсь, что это будет полезно для каждого обучения, и я надеюсь, что все будут поддерживать Wulin.com больше.