2023-03-15 RabbitMQ

RabbitMQ整合

官网erlang版本 : 20.3.8.x

官方rabbitMq版本： rabbitmq-server-generic-unix-3.7.14.tar.xz

1.安装

1.1 安装erlang

1.安装环境

yum -y install make gcc gcc-c++ kernel-devel m4 ncurses-devel openssl-devel

2.在/usr/local/下创建erlangapp文件夹,创建erlang文件夹

mkdir /usr/local/erlangapp
mkdir /usr/local/erlang

2.将erlang安装包上传到该目录

3.解压

tar -zvxf 

4.进入erlang解压目录中

5.由于是configure.in，所以需要使用autoconf生成配置

./otp_build autoconf 

如果报错autoconf找不到，则需要安装

yum -y install autoconf

6.进入文件,编译安装

./configure
make  
make install  

7.配置环境变量

/*进入环境变量配置文件*/
vim /etc/profile
//在底部添加环境变量
export ERLANG_HOME=/usr/local/lib/erlang  
export PATH=$PATH:$ERLANG_HOME/bin 

8.重新加载环境变量

source /etc/profile

9.输入erl查看

erl
halt().//退出

1.2 RabbitMQ安装

1.创建rabbitMQ文件夹，上传

mkdir /usr/local/rabbitmq/

2.解压

 xz -d rabbitmq-server-generic-unix-3.7.14.tar.xz tar -vxf rabbitmq-server-generic-unix-3.7.14.tar 

3.配置环境变量

 vim /etc/profile
向文件添加export PATH=$PATH:/usr/local/rabbitmq/rabbitmq/sbin
//重新加载source /etc/profile

4.启动

rabbitmq-server -detached   【开启rabbitmq服务】
rabbitmqctl status      【查看服务状态】
rabbitmqctl start_app    【启动应用】
rabbitmq-plugins enable rabbitmq_management   【开启web管理插件】

5.添加管理员

rabbitmqctl add_user 用户名 密码
rabbitmqctl set_permissions -p / 用户名 ".*" ".*" ".*"  //配置权限
rabbitmqctl set_user_tags 用户名 administrator  //修改用户角色,将用户设为管理员
rabbitmqctl list_users   //查看所有用户
rabbitmqctl delete_user guest   //删除guest用户

6.启动

http://ip:15672

2.RabbitMQ原生使用

1.pom依赖

	   <dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>3.4.1</version></dependency>

2.连接工具类

package com.li.direct.util;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;import java.io.IOException;public class RabbitMQUtil {private static final String HOST_ADDRESS="192.168.25.102";private static final Integer PORT=5672;private static final String VIRTUAL_HOST="/";private static final String USER_NAME="root";private static final String PASSWORD="root";public static Connection getConnection() throws IOException {com.rabbitmq.client.ConnectionFactory factory=new com.rabbitmq.client.ConnectionFactory();factory.setHost(HOST_ADDRESS);factory.setPort(PORT);factory.setVirtualHost(VIRTUAL_HOST);factory.setUsername(USER_NAME);factory.setPassword(PASSWORD);Connection connection=factory.newConnection();return connection;}public static void main(String[] args) {Connection connection = null;try {connection = getConnection();Channel channel = connection.createChannel();System.out.println(channel);} catch (IOException e) {e.printStackTrace();}finally {try {connection.close();} catch (IOException e) {e.printStackTrace();}}}
}

2.1 消息生产者

//消息提供者
public class MqProvider {//声明交换机public static final String EXCHANGE_NAME="changTest1";//声明队列public static final String QUEUE_NAME="quTest1";public static void main(String[] args) {System.out.println("消息提供者运行。。。");//获取连接Connection connection = null;Channel channel = null;try {connection = RabbitMQUtil.getConnection();//创建信道channel = connection.createChannel();//声明交换机channel.exchangeDeclare(EXCHANGE_NAME,"direct",false,false,false,null);//声明队列channel.queueDeclare(QUEUE_NAME,false,false,false,null);//进行队列绑定channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"test1");String msg = "这是一条消息";int i = 0;//发布消息while (i<100){channel.basicPublish(EXCHANGE_NAME,"test1",null,(msg+i).getBytes(StandardCharsets.UTF_8));i++;}} catch (IOException e) {e.printStackTrace();}finally {if(channel!=null){try {channel.close();} catch (IOException e) {e.printStackTrace();}}if(connection!=null){try {connection.close();} catch (IOException e) {e.printStackTrace();}}}}
}

2.2 消息消费者

//消费测试/*** 当设置自动应答时，如果一个队列有多个消费者，则会平均分配* 如果要按能者多劳，需要设置手动应答，并设置ack流速*/
public class MqConsumer {//声明队列public static final String QUEUE_NAME="quTest1";public static void main(String[] args) {System.out.println("消费者1运行。。。。");//获取连接Connection connection = null;Channel channel = null;try {connection = RabbitMQUtil.getConnection();//创建信道channel = connection.createChannel();//创建消费方QueueingConsumer queueingConsumer = new QueueingConsumer(channel);//控制ack流速,表示每次进行ack确认前只会处理一条消息channel.basicQos(1);//设置自动应答channel.basicConsume(QUEUE_NAME,false,queueingConsumer);//获取消息while (true){QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();byte[] msg = delivery.getBody();//获取消息标识(序号)long tag = delivery.getEnvelope().getDeliveryTag();System.out.println(new String(msg));//消息确认,根据消息序号(false只确认当前一个消息收到，true确认所有比当前序号小的消息（成功消费，消息从队列中删除 ）)channel.basicAck(tag,false);}} catch (Exception e) {e.printStackTrace();}finally {if(channel!=null){try {channel.close();} catch (IOException e) {e.printStackTrace();}}if(connection!=null){try {connection.close();} catch (IOException e) {e.printStackTrace();}}}}
}

3.Spring整合RabbitMQ

pom依赖

   	 <dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit</artifactId><version>1.7.1.RELEASE</version></dependency>

3.1 消息生产者

核心配置文件

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd"><bean id="connectionFactory" class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory"><!-- username,访问RabbitMQ服务器的账户,默认是guest --><property name="username" value="root" /><!-- username,访问RabbitMQ服务器的密码,默认是guest --><property name="password" value="root" /><!-- host,RabbitMQ服务器地址，默认值"localhost" --><property name="host" value="192.168.25.102" /><!-- port，RabbitMQ服务端口，默认值为5672 --><property name="port" value="5672" /></bean><rabbit:admin connection-factory="connectionFactory" /><bean id="jsonMessageConverter"class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter" /><!-- queue 队列声明 --><rabbit:queue  durable="false" auto-delete="false" exclusive="false" name="logQueue" /><rabbit:queue  durable="false" auto-delete="false" exclusive="false" name="NewsQueue" /><!-- exchange queue binging key 绑定 --><rabbit:direct-exchange name="logExchange" durable="false" auto-delete="false"><rabbit:bindings><rabbit:binding queue="logQueue" key="admin:logs" /></rabbit:bindings></rabbit:direct-exchange><rabbit:direct-exchange name="searchExchange" durable="false" auto-delete="false"><rabbit:bindings><rabbit:binding queue="NewsQueue" key="admin:news" /></rabbit:bindings></rabbit:direct-exchange><bean id="rabbitTemplate" class="org.springframework.amqp.rabbit.core.RabbitTemplate"><constructor-arg name="connectionFactory" ref="connectionFactory"/><property name="messageConverter" ref="jsonMessageConverter"/></bean></beans>

消息发送

    @Autowiredprivate AmqpTemplate amqpTemplate;//给消息队列amqpTemplate.convertAndSend("logExchange","admin:logs",logsVo);

3.2 消息消费者

核心配置文件

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd"><bean id="connectionFactory" class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory"><!-- username,访问RabbitMQ服务器的账户,默认是guest --><property name="username" value="root" /><!-- username,访问RabbitMQ服务器的密码,默认是guest --><property name="password" value="root" /><!-- host,RabbitMQ服务器地址，默认值"localhost" --><property name="host" value="192.168.25.102" /><!-- port，RabbitMQ服务端口，默认值为5672 --><property name="port" value="5672" /></bean><rabbit:admin connection-factory="connectionFactory" /><bean id="jsonMessageConverter"class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter" /><bean id="newsSearchConsumerListener" class="com.li.search.listener.NewsSearchConsumerListener"/><!-- queue 队列声明 --><rabbit:queue  durable="false" auto-delete="false" exclusive="false" name="NewsQueue" /><!-- 消息监听 --><rabbit:listener-container connection-factory="connectionFactory" ><rabbit:listener ref="newsSearchConsumerListener" queues="NewsQueue"/></rabbit:listener-container></beans>

消息消费

public class NewsSearchConsumerListener implements MessageListener {@Autowiredprivate ElasticSearchConfig elasticSearchConfig;@Overridepublic void onMessage(Message message) {System.out.println("消息消费。。。。。");//获取消息byte[] msg = message.getBody();try {//反序列化NewsVo newsVo = JSONObject.parseObject(msg, NewsVo.class);//创建要索引的对象NewsIndexVo newsIndexVo = new NewsIndexVo();newsIndexVo.setId(new Long(newsVo.getId()));//添加索引//获取创建的索引IndexRequest indexRequest = new IndexRequest("news");indexRequest.id(newsIndexVo.getId()+"");//HttpRest方式请求，所以请求体必须以http方式进行请求，格式符合http规范String jsonString = JSONObject.toJSONString(newsIndexVo);//设置请求的参数为json数据indexRequest.source(jsonString, XContentType.JSON);//对数据进行索引操作elasticSearchConfig.esRestClient().index(indexRequest,ElasticSearchConfig.COMMON_OPTIONS);} catch (IOException e) {e.printStackTrace();}}
}

4.SpringBoot整合RabbitMQ

pom依赖

	 <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>

1.yml核心配置

 #配置rabbitmq
spring: rabbitmq:host: 192.168.25.102 #主机地址port: 5672  #端口号username: root #用户名password: root #密码virtual-host: /  #虚拟主机地址listener:  #消息监听配置simple:acknowledge-mode: manual #手动确认消息  auto自动确认

2.核心配置类

@Configuration
public class RabbitMQConfig {//声明交换机@Beanpublic DirectExchange newsExchange(){return new DirectExchange("newsExchange");}//声明队列@Beanpublic Queue newsRemoteQueue(){return new Queue("newsRemoteQueue");}//交换机队列绑定@Beanpublic Binding springExchangeBindSpringQueue(){return  BindingBuilder.bind(newsRemoteQueue()).to(newsExchange()).with("admin:news");}}

3.消息队列监听

@Component
public class RabbitMQListener {@Autowiredprivate NewsMapper newsMapper;@Autowiredprivate NewsRemoteNewsMapper newsRemoteNewsMapper;//将新闻数据存到新闻表@RabbitListener(queues = "xxx")@RabbitHandlerpublic void addNews(Channel channel, Message message){String newsStr = new String(message.getBody());NewsRemoteNews newsRemoteNews = JSONObject.parseObject(newsStr, NewsRemoteNews.class);}}

5.常见场景应用

5.1 死信队列

消息生产

//死信
public class DlxProvider {//声明死信队列public static final String DLX_EXCHANGE_NAME = "dlx_exchange";public static final String DLX_QUEUE_NAME="dlx_queue";//声明普通队列public static final String EXCHANGE_NAME = "exchange_test";public static final String QUEUE_NAME = "queue_test";public static void main(String[] args) throws Exception {Connection connection = RabbitMQUtil.getConnection();Channel channel = connection.createChannel();//声明死信交换机channel.exchangeDeclare(DLX_EXCHANGE_NAME,"direct",false,false,false,null);//声明死信队列channel.queueDeclare(DLX_QUEUE_NAME,false,false,false,null);//绑定死信队列channel.queueBind(DLX_QUEUE_NAME,DLX_EXCHANGE_NAME,"");//声明普通交换机channel.exchangeDeclare(EXCHANGE_NAME,"direct",false,false,false,null);//设置参数Map<String,Object> map = new HashMap<>();//map.put("x-message-ttl",6000);//超时时间map.put("x-dead-letter-exchange",DLX_EXCHANGE_NAME);//要去的死信队列map.put("x-dead-letter-routing-key","");//死信队列路由key//声明队列channel.queueDeclare(QUEUE_NAME,false,false,false,map);//绑定队列channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"abc");//发送消息channel.basicPublish(EXCHANGE_NAME,"abc",null,"这是一条消息".getBytes());channel.close();connection.close();}
}

死信队列

//死信队列消费者
public class DlxQueueConsumer {public static final String DLX_QUEUE_NAME="dlx_queue";public static void main(String[] args) throws Exception {System.out.println("死信队列消费运行。。。。");//获取连接Connection connection = RabbitMQUtil.getConnection();//创建信道Channel channel = connection.createChannel();//创建消费方QueueingConsumer queueingConsumer = new QueueingConsumer(channel);//控制ack流速,表示每次进行ack确认前只会处理一条消息channel.basicQos(1);//设置自动应答channel.basicConsume(DLX_QUEUE_NAME, true, queueingConsumer);while (true){QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();//获取消息byte[] msg = delivery.getBody();//获取消息标识(序号)long tag = delivery.getEnvelope().getDeliveryTag();System.out.println(new String(msg));}}
}

死信消费者

//消费者（测试死信队列）
public class DlxConsumer {//声明队列public static final String QUEUE_NAME = "queue_test";public static void main(String[] args) throws Exception {System.out.println("消费者运行。。。。");//获取连接Connection connection = RabbitMQUtil.getConnection();//创建信道Channel channel = connection.createChannel();//创建消费方QueueingConsumer queueingConsumer = new QueueingConsumer(channel);//控制ack流速,表示每次进行ack确认前只会处理一条消息channel.basicQos(1);//设置自动应答channel.basicConsume(QUEUE_NAME, false, queueingConsumer);QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();//获取消息byte[] msg = delivery.getBody();//获取消息标识(序号)long tag = delivery.getEnvelope().getDeliveryTag();try {System.out.println(new String(msg));//业务正常时，消息确认channel.basicAck(tag, false);}catch (Exception e){//如果业务出现异常，服务器异常，拒绝确认消息,第三个参数代表是否重新入队//拒绝重新入队和拒绝不确认时，会进入死信队列，//如果允许重新入队则需保证rabbit集群，否则会死循环channel.basicNack(tag,false,false);}channel.close();connection.close();}
}

5.2 延迟消息处理

第一种是使用普通队列和死信队列来模拟实现延迟的效果。将消息放入一个没有被监听的队列上，设置TTL(一条消息的最大存活时间)为延迟的时间，时间到了没有被消费，直接成为死信，进入死信队列。后监听私信队列来消息消费

第二种是使用rabbitmq官方提供的delayed插件来真正实现延迟队列。