SpringBoot整合RabbitMQ,實現訊息傳送和消費
阿新 • • 發佈:2018-12-05
下載安裝Erlang和RabbitMQ
Erlang和RabbitMQ:https://www.cnblogs.com/theRhyme/p/10069611.html
專案建立和依賴
推薦SpringCloud專案線上建立:https://start.spring.io/
不用上面這個也行,下面有程式碼和依賴;
gradle的依賴,和maven差不多:
buildscript { ext { springBootVersion = '2.1.1.RELEASE' } repositories { mavenCentral() } dependencies { classpath("org.springframework.boot:spring-boot-gradle-plugin:${springBootVersion}") } } apply plugin: 'java' apply plugin: 'eclipse' apply plugin: 'org.springframework.boot' apply plugin: 'io.spring.dependency-management' group = 'xy.study' version = '0.0.1-SNAPSHOT' sourceCompatibility = 1.8 repositories { mavenCentral() } dependencies { implementation('org.springframework.boot:spring-boot-starter-amqp') implementation('org.projectlombok:lombok:1.16.+') runtimeOnly('org.springframework.boot:spring-boot-devtools') testImplementation('org.springframework.boot:spring-boot-starter-test') }
程式碼
配置檔案application.properties
spring.application.name=spring-boot-rabbitmq spring.rabbitmq.host=localhost spring.rabbitmq.port=5672 spring.rabbitmq.username=guest spring.rabbitmq.password=guest server.port = 5678
RabbitMQ配置檔案類(註釋的程式碼可以直接刪掉):
import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.Queue; import org.springframework.amqp.core.TopicExchange; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; /** * topic 是RabbitMQ中最靈活的一種方式,可以根據routing_key自由的繫結不同的佇列 * 首先對topic規則配置 */ //@Configuration public class TopicRabbitConfig { final public static String QUEUE_NAME = "queue.name"; final public static String TEST_TOPIC_ROUTINGKEY = "test.topic.routingKey"; final public static String TEST_EXCHANGE_HAHA = "test.exchange.haha"; /** * 設定交換器的名稱 * @return *//* @Bean TopicExchange exchange() { return new TopicExchange(TopicRabbitConfig.TEST_EXCHANGE_HAHA); } *//** * 佇列名稱 * @return *//* @Bean public Queue queueMessage() { return new Queue(TopicRabbitConfig.QUEUE_NAME); } *//** * 將指定routing key的名稱繫結交換器的佇列 * @param queueMessage * @param exchange * @return *//* @Bean Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) { return BindingBuilder.bind(queueMessage).to(exchange).with(TopicRabbitConfig.TEST_TOPIC_ROUTINGKEY); }*/ /** * 匹配以topic開頭的路由鍵routing key * 交換機繫結多個佇列 */ /*@Bean Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) { return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#"); }*/ }
生產者,這裡根據Exchange和Routing Key,直接傳送一個字串:
import lombok.extern.slf4j.Slf4j; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import xy.study.rabbitmq.conf.TopicRabbitConfig; @Component @Slf4j public class HelloSender { @Autowired private RabbitTemplate rabbitTemplate; /** * 通過exchange和routingKey的方式 * rabbitTemplate.convertAndSend(String exchange, String routingKey, Object object) * @param i */ public void send(int i) { String context = "hello " + i; log.info("Sender : {}", context); this.rabbitTemplate.convertAndSend(TopicRabbitConfig.TEST_EXCHANGE_HAHA,TopicRabbitConfig.TEST_TOPIC_ROUTINGKEY, context); } }
消費者,繫結對應的Exchange,Queue和Routing Key,直接列印獲取的資訊:
import lombok.extern.slf4j.Slf4j; import org.springframework.amqp.core.ExchangeTypes; import org.springframework.amqp.rabbit.annotation.Exchange; import org.springframework.amqp.rabbit.annotation.Queue; import org.springframework.amqp.rabbit.annotation.QueueBinding; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.messaging.handler.annotation.Payload; import org.springframework.stereotype.Component; import xy.study.rabbitmq.conf.TopicRabbitConfig; @Component @Slf4j public class HelloReceiver { @RabbitListener(bindings = @QueueBinding( value = @Queue(value = TopicRabbitConfig.QUEUE_NAME, durable = "true"), exchange = @Exchange(value = TopicRabbitConfig.TEST_EXCHANGE_HAHA, type = ExchangeTypes.TOPIC), key = TopicRabbitConfig.TEST_TOPIC_ROUTINGKEY) ) public void onOrgDeleted(@Payload String msg) { log.info("HelloReceiver msg : {}",msg); } }
測試類,呼叫生產者傳送資訊的函式send,消費者會監聽消費:
import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringRunner; import xy.study.rabbitmq.producer.HelloSender; @RunWith(SpringRunner.class) @SpringBootTest public class RabbitmqApplicationTests { @Autowired private HelloSender sender; @Test public void testSend() { sender.send(666); } }
如圖,控制檯日誌,能生成訊息,並且能被對應的消費者消費。