详解spring boot集成RabbitMQ

详解spring boot集成RabbitMQ

RabbitMQ作为AMQP的代表性产品，在项目中大量使用。结合现在主流的spring boot，极大简化了开发过程中所涉及到的消息通信问题。

首先正确的安装RabbitMQ及运行正常。

RabbitMQ需啊erlang环境，所以首先安装对应版本的erlang，可在RabbitMQ官网下载

# rpm -ivh erlang-19.0.4-1.el7.centos.x86_64.rpm

使用yum安装RabbitMQ,避免缺少依赖包引起的安装失败

# yum install rabbitmq-server-3.6.6-1.el7.noarch.rpm

启动RabbitMQ

# /sbin/service rabbitmq-server start

由于RabbitMQ默认提供的guest用户只能本地访问，所以额外创建用户用于测试

# /sbin/rabbitmqctl add_user test test123

用户名：test，密码：test123

开启web管理插件

# rabbitmq-plugins enable rabbitmq_management

并使用之前创建的用户登录,并设置该用户为administrator,虚拟主机地址为/

spring boot 引入相关依赖

org.springframework.boot

spring-boot-starter-amqp

消息生产者

application.properties添加一下配置

spring.rabbitmq.host=192.168.1.107

spring.rabbitmq.port=5672

spring.rabbitmq.username=test

spring.rabbitmq.password=test123

spring.rabbitmq.publisher-confirms=true

spring.rabbitmq.publisher-returns=true

spring.rabbitmq.template.mandatory=true

spring boot配置类，作用为指定队列，交换器类型及绑定操作

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;

@Configuration

public class RabbitConfig {

//声明队列

@Bean

public Queue queue1() {

return new Queue("hello.queue1", true); // true表示持久化该队列

}

@Bean

public Queue queue2() {

return new Queue("hello.queue2", true);

}

//声明交互器

@Bean

TopicExchange topicExchange() {

return new TopicExchange("topicExchange");

}

//绑定

@Bean

public Binding binding1() {

return BindingBuilder.bind(queue1()).to(topicExchange()).with("key.1");

}

@Bean

public Binding binding2() {

return BindingBuilder.bind(queue2()).to(topicExchange()).with("key.#");

}

}

共声明了2个队列，分别是hello.queue1，hello.queue2，交换器类型为TopicExchange,并与hello.queue1，hello.queue2队列分别绑定。

生产者类

import java.util.UUID;

import javax.annotation.PostConstruct;

import org.springframework.amqp.core.Message;

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;

import org.springframework.amqp.rabbit.support.CorrelationData;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Component;

@Component

public class Sender implements RabbitTemplate.ConfirmCallback, ReturnCallback {

@Autowired

private RabbitTemplate rabbitTemplate;

@PostConstruct

public void init() {

rabbitTemplate.setConfirmCallback(this);

rabbitTemplate.setReturnCallback(this);

}

@Override

public void confirm(CorrelationData correlationData, boolean ack, String cause) {

if (ack) {

System.out.println("消息发送成功:" + correlationData);

} else {

System.out.println("消息发送失败:" mWdaWqWUU+ cause);

}

}

@Override

public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {

System.out.println(message.getMessageProperties().getCorrelationIdString() + " mWdaWqWUU发送失败");

}

//发送消息，不需要实现任何接口，供外部调用。

public void send(String msg){

CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());

System.out.println("开始发送消息 : " + msg.toLowerCase());

String response = rabbitTemplate.convertSendAndReceive("topicExchange", "key.1", msg, correlationId).toString();

System.out.println("结束发送消息 : " + msg.toLowerCase());

System.out.println("消费者响应 : " + response + " 消息处理完成");

}

}

要点：

1.注入RabbitTemplate

2.实现RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback接口（非必须）。

ConfirmCallback接口用于实现消息发送到RabbitMQ交换器后接收ack回调。ReturnCallback接口用于实现消息发送到RabbitMQ交换器，但无相应队列与交换器绑定时的回调。

3.实现消息发送方法。调用rabbitTemplatehttp://相应的方法即可。rabbitTemplate常用发送方法有

rabbitTemplate.send(message); //发消息，参数类型为org.springframework.amqp.core.Message

rabbitTemplate.convertAndSend(object); //转换并发送消息。 将参数对象转换为org.springframework.amqp.core.Message后发送

rabbitTemplate.convertSendAndReceive(message) //转换并发送消息,且等待消息者返回响应消息。

针对业务场景选择合适的消息发送方式即可。

消息消费者

application.properties添加一下配置

spring.rabbitmq.host=192.168.1.107

spring.rabbitmq.port=5672

spring.rabbitmq.username=test

spring.rabbitmq.password=test123

spring.rabbitmq.listener.concurrency=2 //最小消息监听线程数

spring.rabbitmq.listener.max-concurrency=2 //最大消息监听线程数

消费者类

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.stereotype.Component;

@Component

public class Receiver {

@RabbitListener(queues = "hello.queue1")

public String processMessage1(String msg) {

System.out.println(Thread.currentThread().getName() + " 接收到来自hello.queue1队列的消息：" + msg);

return msg.toUpperCase();

}

@RabbitListener(queues = "hello.queue2")

public void processMessage2(String msg) {

System.out.println(Thread.currentThread().getName() + " 接收到来自hello.queue2队列的消息：" + msg);

}

}

由于定义了2个队列，所以分别定义不同的监听器监听不同的队列。由于最小消息监听线程数和最大消息监听线程数都是2，所以每个监听器各有2个线程实现监听功能。

要点:

1.监听器参数类型与消息实际类型匹配。在生产者中发送的消息实际类型是String，所以这里监听器参数类型也是String。

2.如果监听器需要有响应返回给生产者，直接在监听方法中return即可。

运行测试

import java.util.Date;

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.SpringJUnit4ClassRunner;

import com.sam.demo.rabbitmq.Application;

import com.sam.demo.rabbitmq.sender.Sender;

@RunWith(value=SpringJUnit4ClassRunner.class)

@SpringBootTest(classes = Application.class)

public class RabbitTests {

@Autowired

private Sender sender;

@Test

public void sendTest() throws Exception {

while(true){

String msg = new Date().toString();

sender.send(msg);

Thread.sleep(1000);

}

}

}

输出:

开始发送消息 : wed mar 29 23:20:52 cst 2017

SimpleAsyncTaskExecutor-1 接收到来自hello.queue2队列的消息：Wed Mar 29 23:20:52 CST 2017

SimpleAsyncTaskExecutor-2 接收到来自hello.queue1队列的消息：Wed Mar 29 23:20:52 CST 2017

结束发送消息 : wed mar 29 23:20:52 cst 2017

消费者响应 : WED MAR 29 23:20:52 CST 2017 消息处理完成

------------------------------------------------

消息发送成功:CorrelationData [id=340d14e6-cfcc-4653-9f95-29b37d50f886]

开始发送消息 : wed mar 29 23:20:53 cst 2017

SimpleAsyncTaskExecutor-1 接收到来自hello.queue1队列的消息：Wed Mar 29 23:20:53 CST 2017

SimpleAsyncTaskExecutor-2 接收到来自hello.queue2队列的消息：Wed Mar 29 23:20:53 CST 2017

结束发送消息 : wed mar 29 23:20:53 cst 2017

消费者响应 : WED MAR 29 23:20:53 CST 2017 消息处理完成

------------------------------------------------

消息发送成功:CorrelationData [id=e4e01f89-d0d4-405e-80f0-85bb20238f34]

开始发送消息 : wed mar 29 23:20:54 cst 2017

SimpleAsyncTaskExecutor-2 接收到来自hello.queue1队列的消息：Wed Mar 29 23:20:54 CST 2017

SimpleAsyncTaskExecutor-1 接收到来自hello.queue2队列的消息：Wed Mar 29 23:20:54 CST 2017

结束发送消息 : wed mar 29 23:20:54 cst 2017

消费者响应 : WED MAR 29 23:20:54 CST 2017 消息处理完成

------------------------------------------------

如果需要使用的其他的交换器类型，spring中都已提供实现，所有的交换器均实现org.springframework.amqp.core.AbstractExchange接口。

常用交换器类型如下：

Direct(DirectExchange)：direct 类型的行为是"先匹配, 再投送". 即在绑定时设定一个 routing_key, 消息的routing_key完全匹配时, 才会被交换器投送到绑定的队列中去。

Topic(TopicExchange)：按规则转发消息（最灵活）。

Headers(HeadersExchange)：设置header attribute参数类型的交换机。

Fanout(FanoutExchange)：转发消息到所有绑定队列。

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。