本篇主要rabbitMq相关的一些简单理论介绍，安装须知，provider消息推送实例，consumer消息消费实例，Direct、Topic、Fanout的使用。
下一篇会讲解消息回调、手动确认等。
安装主要需要几个介质，Linux环境上 需安装erlang、 rabbitmqserver、socat 大家根据版本下载不要一定要安装对应版本否则无法启动。
Erlang 下载地址：
https://www.erlang-solutions.com/resources/download.html
安装完软件后进行配置：
1 配置文件
vim /etc/security/limits.conf 新增
* soft nofile 65535
* hard nofile 65535
2.配置配置文件，需要先将文件拷贝到/etc/rabbitmq/ 下然后重新命名一下。启动时会需要扫描该文件，可以在文件里设置ip 监听端口，默认帐号，开启远程访问等等。。。
cd /etc/rabbitmq
cp /usr/share/doc/rabbitmq-server-3.6.12/rabbitmq.config.example /etc/rabbitmq/
mv rabbitmq.config.example rabbitmq.config
设置开机自启动:
chkconfig rabbitmq-server on
3.开启用户远程访问
vi /etc/rabbitmq/rabbitmq.config
注意要去掉后面的逗号。改成下图
4.开启web界面管理工具
rabbitmq-plugins enable rabbitmq_management
service rabbitmq-server restart
5.启停RabbitMQ 服务
service rabbitmq-server start
service rabbitmq-server stop
service rabbitmq-server restart
6.切记开放端口:15672/5672
进如WEBUI界面进行后续配置，默认账户guest/guest
7.添加Virtual Hosts
可以先添加用户
第一个是我安装的rabbitmq版本第二个是erlang版本 3.6只能用erlang1.9.0或以下的版本
添加用户tags 可以设置这个用户的权限。
添加完Virtual Hosts 后可以使用用户何其进行关联
在这个页面控制台上我们可以做些什么？
可以手动创建虚拟host，创建用户，分配权限，创建交换机，创建队列等等，还有查看队列消息，消费效率，推送效率等等。暂时我不截图做讲解等后续大家安装后点击看看基本都能理解。
8.交换机种类
- Direct Exchange
直连型交换机，根据消息携带的路由键将消息投递给对应队列。
大致流程，有一个队列绑定到一个直连交换机上，同时赋予一个路由键 Routing Key 。
然后当一个消息携带着路由值为A，这个消息通过生产者发送给交换机时，交换机就会根据这个路由值A去寻找绑定值也是A的队列。
- Fanout Exchange
扇型交换机，这个交换机没有路由键概念，就算你绑了路由键也是无视的。 这个交换机在接收到消息后，会直接转发到绑定到它上面的所有队列。
- Topic Exchange
主题交换机，这个交换机其实跟直连交换机流程差不多，但是它的特点就是在它的路由键和绑定键之间是有规则的。
介绍下规则：
* (星号) 用来表示一个单词 (必须出现的)
# (井号) 用来表示任意数量（零个或多个）单词
通配的绑定键是跟队列进行绑定的，举个小例子
队列Q1 绑定键为 *.AA.* 队列Q2绑定键为 AA.#
如果一条消息携带的路由键为 A.CC.B，那么队列Q1将会收到；
如果一条消息携带的路由键为CC.AA.BB，那么队列Q2将会收到；
主题交换机是非常强大的，为啥这么膨胀？
当一个队列的绑定键为 "#"（井号） 的时候，这个队列将会无视消息的路由键，接收所有的消息。
当 * (星号) 和 # (井号) 这两个特殊字符都未在绑定键中出现的时候，此时主题交换机就拥有的直连交换机的行为。
所以主题交换机也就实现了扇形交换机的功能，和直连交换机的功能。
除了上述三种还有 Header Exchange 头交换机 ，Default Exchange 默认交换机，Dead Letter Exchange 死信交换机。
简单说以下死信交换机，有些公司会使用私信交换机进行消息延迟发送，或者将一些无法消费的消息移动到私信交换机进行再次处理。
9 编码
1 本教程创建2个springboot项目，一个 rabbitmq-provider （生产者），一个rabbitmq-consumer（消费者）。
2 生产者和消费者都添加配置文件application.yml ，根据自己服务器设置的ip端口帐号进行填写即可
-
server:
-
port:
8021
-
spring:
-
#给项目来个名字
-
application:
-
name: rabbitmq-provider
-
#配置rabbitMq 服务器
-
rabbitmq:
-
host:
192.168.3.32
-
port:
5672
-
username: admin
-
password: admin
-
#虚拟host 可以不设置,使用server默认host
-
#virtual-host: Test
3 pom.xml 添加依赖 生成者和消费者都要添加
-
<!--rabbitmq-->
-
<dependency>
-
<groupId>org.springframework.boot</groupId>
-
<artifactId>spring-boot-starter-amqp</artifactId>
-
</dependency>
10 我们先从生产者开始写 首先生产者里创建 RabbitConfig 队列定义和SendMessageController 接口
-
package com.provider.rabbitmqprovider;
-
-
import org.springframework.amqp.core.*;
-
import org.springframework.context.annotation.Bean;
-
-
import org.springframework.context.annotation.Configuration;
-
-
-
/**
-
* @Title: RabbitMQ
-
* @Description:
-
* @author: liaryank
-
* @Date: 2020/5/14 3:10 下午
-
* @Version: 1.0
-
*/
-
-
@Configuration
-
public
class RabbitConfig {
-
-
//绑定键
-
public
final
static String liaryank =
"topic.liaryank";
-
public
final
static String liar =
"topic.liar";
-
-
-
//队列 起名：TestDirectQueue
-
@Bean
-
public Queue TestDirectQueue() {
-
// durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效
-
// exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable
-
// autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。
-
// return new Queue("TestDirectQueue",true,true,false);
-
-
//一般设置一下队列的持久化就好,其余两个就是默认false
-
return
new Queue(
"TestDirectQueue",
true);
-
}
-
-
//Direct交换机 起名：TestDirectExchange
-
@Bean
-
DirectExchange TestDirectExchange() {
-
// return new DirectExchange("TestDirectExchange",true,true);
-
return
new DirectExchange(
"TestDirectExchange",
true,
false);
-
}
-
-
//绑定 将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting
-
@Bean
-
Binding bindingDirect() {
-
Binding testDirectRouting = BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with(
"TestDirectRouting");
-
return testDirectRouting;
-
}
-
-
@Bean
-
DirectExchange lonelyDirectExchange() {
-
return
new DirectExchange(
"lonelyDirectExchange");
-
}
-
-
// topic 类型
-
@Bean
-
public Queue firstQueue() {
-
return
new Queue(RabbitConfig.liaryank);
-
}
-
-
@Bean
-
public Queue secondQueue() {
-
return
new Queue(RabbitConfig.liar);
-
}
-
-
@Bean
-
TopicExchange exchange() {
-
return
new TopicExchange(
"topicExchange");
-
}
-
-
-
//将firstQueue和topicExchange绑定,而且绑定的键值为topic.man
-
//这样只要是消息携带的路由键是topic.man,才会分发到该队列
-
@Bean
-
Binding bindingExchangeMessage() {
-
return BindingBuilder.bind(firstQueue()).to(exchange()).with(liaryank);
-
}
-
-
//将secondQueue和topicExchange绑定,而且绑定的键值为用上通配路由键规则topic.#
-
// 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列
-
@Bean
-
Binding bindingExchangeMessage2() {
-
return BindingBuilder.bind(secondQueue()).to(exchange()).with(
"topic.#");
-
}
-
-
/**
-
* 创建三个队列 ：fanout.A fanout.B fanout.C
-
* 将三个队列都绑定在交换机 fanoutExchange 上
-
* 因为是扇型交换机, 路由键无需配置,配置也不起作用
-
*/
-
@Bean
-
public Queue queueA() {
-
return
new Queue(
"fanout.A");
-
}
-
-
@Bean
-
public Queue queueB() {
-
return
new Queue(
"fanout.B");
-
}
-
-
@Bean
-
public Queue queueC() {
-
return
new Queue(
"fanout.C");
-
}
-
-
@Bean
-
FanoutExchange fanoutExchange() {
-
return
new FanoutExchange(
"fanoutExchange");
-
}
-
-
@Bean
-
Binding bindingExchangeA() {
-
return BindingBuilder.bind(queueA()).to(fanoutExchange());
-
}
-
-
@Bean
-
Binding bindingExchangeB() {
-
return BindingBuilder.bind(queueB()).to(fanoutExchange());
-
}
-
-
@Bean
-
Binding bindingExchangeC() {
-
return BindingBuilder.bind(queueC()).to(fanoutExchange());
-
}
-
}
-
package com.provider.rabbitmqprovider;
-
-
import org.springframework.amqp.rabbit.core.RabbitTemplate;
-
import org.springframework.beans.factory.annotation.Autowired;
-
import org.springframework.web.bind.annotation.GetMapping;
-
import org.springframework.web.bind.annotation.RestController;
-
-
import java.time.LocalDateTime;
-
import java.time.format.DateTimeFormatter;
-
import java.util.HashMap;
-
import java.util.Map;
-
import java.util.UUID;
-
-
/**
-
* @Title: RabbitMQ
-
* @Description:
-
* @author: liaryank
-
* @Date: 2020/5/14 3:17 下午
-
* @Version: 1.0
-
*/
-
-
@RestController
-
public
class SendMessageController {
-
@Autowired
-
RabbitTemplate rabbitTemplate;
//使用RabbitTemplate,这提供了接收/发送等等方法
-
-
//Direct类型
-
@GetMapping(
"/sendDirectMessage")
-
public String sendDirectMessage() {
-
String messageId = String.valueOf(UUID.randomUUID());
-
String messageData =
"message is hello!";
-
String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern(
"yyyy-MM-dd HH:mm:ss"));
-
Map<String,Object> map=
new HashMap<>();
-
map.put(
"messageId",messageId);
-
map.put(
"messageData",messageData);
-
map.put(
"createTime",createTime);
-
//将消息携带绑定键值：TestDirectRouting 发送到交换机TestDirectExchange
-
rabbitTemplate.convertAndSend(
"TestDirectExchange",
"TestDirectRouting", map);
-
return
"ok";
-
}
-
-
//Topic类型
-
@GetMapping(
"/sendTopicMessage1")
-
public String sendTopicMessage1() {
-
String messageId = String.valueOf(UUID.randomUUID());
-
String messageData =
"message: liaryank~~~ ";
-
String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern(
"yyyy-MM-dd HH:mm:ss"));
-
Map<String, Object> manMap =
new HashMap<>();
-
manMap.put(
"messageId", messageId);
-
manMap.put(
"messageData", messageData);
-
manMap.put(
"createTime", createTime);
-
rabbitTemplate.convertAndSend(
"topicExchange",
"topic.liaryank", manMap);
-
return
"ok";
-
}
-
-
@GetMapping(
"/sendTopicMessage2")
-
public String sendTopicMessage2() {
-
String messageId = String.valueOf(UUID.randomUUID());
-
String messageData =
"message: msg is all ";
-
String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern(
"yyyy-MM-dd HH:mm:ss"));
-
Map<String, Object> womanMap =
new HashMap<>();
-
womanMap.put(
"messageId", messageId);
-
womanMap.put(
"messageData", messageData);
-
womanMap.put(
"createTime", createTime);
-
rabbitTemplate.convertAndSend(
"topicExchange",
"topic.liar", womanMap);
-
return
"ok";
-
}
-
-
//Fanout类型
-
@GetMapping(
"/sendFanoutMessage")
-
public String sendFanoutMessage() {
-
String messageId = String.valueOf(UUID.randomUUID());
-
String messageData =
"message: testFanoutMessage ";
-
String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern(
"yyyy-MM-dd HH:mm:ss"));
-
Map<String, Object> map =
new HashMap<>();
-
map.put(
"messageId", messageId);
-
map.put(
"messageData", messageData);
-
map.put(
"createTime", createTime);
-
rabbitTemplate.convertAndSend(
"fanoutExchange",
null, map);
-
return
"ok";
-
}
-
-
}
生产者和接口就定义完成我们来继续写消费者
11 三种类型交换机和其对应的消费
12 Direct类型
-
package com.consumer.rabbitmqconsumer;
-
-
-
import org.springframework.amqp.core.Binding;
-
import org.springframework.amqp.core.BindingBuilder;
-
import org.springframework.amqp.core.DirectExchange;
-
import org.springframework.amqp.core.Queue;
-
import org.springframework.context.annotation.Bean;
-
import org.springframework.context.annotation.Configuration;
-
-
/**
-
* @Title: RabbitMQ
-
* @Description:
-
* @author: liaryank
-
* @Date: 2020/5/14 4:01 下午
-
* @Version: 1.0
-
*/
-
-
@Configuration
-
public
class DirectRabbitConfig {
-
//队列 起名：TestDirectQueue
-
@Bean
-
public Queue TestDirectQueue() {
-
return
new Queue(
"TestDirectQueue",
true);
-
}
-
-
//Direct交换机 起名：TestDirectExchange
-
@Bean
-
DirectExchange TestDirectExchange() {
-
return
new DirectExchange(
"TestDirectExchange");
-
}
-
-
//绑定 将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting
-
@Bean
-
Binding bindingDirect() {
-
return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with(
"TestDirectRouting");
-
}
-
-
}
-
package com.consumer.rabbitmqconsumer;
-
-
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
-
import org.springframework.amqp.rabbit.annotation.RabbitListener;
-
import org.springframework.stereotype.Component;
-
-
import java.util.Map;
-
-
/**
-
* @Title: RabbitMQ
-
* @Description:
-
* @author: liaryank
-
* @Date: 2020/5/14 4:03 下午
-
* @Version: 1.0
-
*/
-
@Component
-
@RabbitListener(queues =
"TestDirectQueue")
//监听的队列名称 TestDirectQueue
-
public
class DirectReceiver {
-
@RabbitHandler
-
public void process(Map testMessage) {
-
System.out.println(
"DirectReceiver消费者收到消息 : " + testMessage.toString());
-
}
-
-
}
13 FanoutRabbit类型
-
package com.consumer.rabbitmqconsumer;
-
-
import org.springframework.amqp.core.Binding;
-
import org.springframework.amqp.core.BindingBuilder;
-
import org.springframework.amqp.core.FanoutExchange;
-
import org.springframework.amqp.core.Queue;
-
import org.springframework.context.annotation.Bean;
-
import org.springframework.context.annotation.Configuration;
-
-
/**
-
* @Title: RabbitMQ
-
* @Description:
-
* @author: liaryank
-
* @Date: 2020/5/14 4:38 下午
-
* @Version: 1.0
-
*/
-
-
@Configuration
-
public
class FanoutRabbitConfig {
-
-
/**
-
* 创建三个队列 ：fanout.A fanout.B fanout.C
-
* 将三个队列都绑定在交换机 fanoutExchange 上
-
* 因为是扇型交换机, 路由键无需配置,配置也不起作用
-
*/
-
-
-
@Bean
-
public Queue queueA() {
-
return
new Queue(
"fanout.A");
-
}
-
-
@Bean
-
public Queue queueB() {
-
return
new Queue(
"fanout.B");
-
}
-
-
@Bean
-
public Queue queueC() {
-
return
new Queue(
"fanout.C");
-
}
-
-
@Bean
-
FanoutExchange fanoutExchange() {
-
return
new FanoutExchange(
"fanoutExchange");
-
}
-
-
@Bean
-
Binding bindingExchangeA() {
-
return BindingBuilder.bind(queueA()).to(fanoutExchange());
-
}
-
-
@Bean
-
Binding bindingExchangeB() {
-
return BindingBuilder.bind(queueB()).to(fanoutExchange());
-
}
-
-
@Bean
-
Binding bindingExchangeC() {
-
return BindingBuilder.bind(queueC()).to(fanoutExchange());
-
}
-
}
-
package com.consumer.rabbitmqconsumer;
-
-
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
-
import org.springframework.amqp.rabbit.annotation.RabbitListener;
-
import org.springframework.stereotype.Component;
-
-
import java.util.Map;
-
-
/**
-
* @Title: RabbitMQ
-
* @Description:
-
* @author: liaryank
-
* @Date: 2020/5/14 4:36 下午
-
* @Version: 1.0
-
*/
-
@Component
-
@RabbitListener(queues =
"fanout.A")
-
public
class FanoutReceiverA {
-
-
@RabbitHandler
-
public void process(Map testMessage) {
-
System.out.println(
"FanoutReceiverA消费者收到消息 : " +testMessage.toString());
-
}
-
-
}
-
package com.consumer.rabbitmqconsumer;
-
-
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
-
import org.springframework.amqp.rabbit.annotation.RabbitListener;
-
import org.springframework.stereotype.Component;
-
-
import java.util.Map;
-
-
/**
-
* @Title: RabbitMQ
-
* @Description:
-
* @author: liaryank
-
* @Date: 2020/5/14 4:36 下午
-
* @Version: 1.0
-
*/
-
@Component
-
@RabbitListener(queues =
"fanout.B")
-
public
class FanoutReceiverB {
-
-
@RabbitHandler
-
public void process(Map testMessage) {
-
System.out.println(
"FanoutReceiverB消费者收到消息 : " +testMessage.toString());
-
}
-
-
}
-
package com.consumer.rabbitmqconsumer;
-
-
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
-
import org.springframework.amqp.rabbit.annotation.RabbitListener;
-
import org.springframework.stereotype.Component;
-
-
import java.util.Map;
-
-
/**
-
* @Title: RabbitMQ
-
* @Description:
-
* @author: liaryank
-
* @Date: 2020/5/14 4:36 下午
-
* @Version: 1.0
-
*/
-
@Component
-
@RabbitListener(queues =
"fanout.C")
-
public
class FanoutReceiverC {
-
-
@RabbitHandler
-
public void process(Map testMessage) {
-
System.out.println(
"FanoutReceiverC消费者收到消息 : " +testMessage.toString());
-
}
-
-
}
14 Topic类型
-
package com.consumer.rabbitmqconsumer;
-
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
-
import org.springframework.amqp.rabbit.annotation.RabbitListener;
-
import org.springframework.stereotype.Component;
-
import java.util.Map;
-
/**
-
* @Title: RabbitMQ
-
* @Description: topic
-
* @author: liaryank
-
* @Date: 2020/5/14 4:25 下午
-
* @Version: 1.0
-
*/
-
@Component
-
@RabbitListener(queues =
"topic.liaryank")
-
public
class TopicManReceiver {
-
-
@RabbitHandler
-
public void process(Map testMessage) {
-
System.out.println(
"TopicManReceiver消费者收到消息 : " + testMessage.toString());
-
}
-
}
-
package com.consumer.rabbitmqconsumer;
-
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;
-
/**
-
* @Title: RabbitMQ
-
* @Description:
-
* @author: liaryank
-
* @Date: 2020/5/14 4:27 下午
-
* @Version: 1.0
-
*/
-
@Configuration
-
public
class TopicRabbitConfig {
-
//绑定键
-
public
final
static String liaryank =
"topic.liaryank";
-
public
final
static String liar =
"topic.liar";
-
-
@Bean
-
public Queue firstQueue() {
-
return
new Queue(TopicRabbitConfig.liaryank);
-
}
-
-
@Bean
-
public Queue secondQueue() {
-
return
new Queue(TopicRabbitConfig.liar);
-
}
-
-
@Bean
-
TopicExchange exchange() {
-
return
new TopicExchange(
"topicExchange");
-
}
-
-
-
//将firstQueue和topicExchange绑定,而且绑定的键值为topic.man
-
//这样只要是消息携带的路由键是topic.man,才会分发到该队列
-
@Bean
-
Binding bindingExchangeMessage() {
-
return BindingBuilder.bind(firstQueue()).to(exchange()).with(liaryank);
-
}
-
-
//将secondQueue和topicExchange绑定,而且绑定的键值为用上通配路由键规则topic.#
-
// 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列
-
@Bean
-
Binding bindingExchangeMessage2() {
-
return BindingBuilder.bind(secondQueue()).to(exchange()).with(
"topic.#");
-
}
-
-
}
-
package com.consumer.rabbitmqconsumer;
-
-
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
-
import org.springframework.amqp.rabbit.annotation.RabbitListener;
-
import org.springframework.stereotype.Component;
-
import java.util.Map;
-
/**
-
* @Title: RabbitMQ
-
* @Description:
-
* @author: liaryank
-
* @Date: 2020/5/14 4:26 下午
-
* @Version: 1.0
-
*/
-
@Component
-
@RabbitListener(queues =
"topic.liar")
-
public
class TopicTotalReceiver {
-
@RabbitHandler
-
public void process(Map testMessage) {
-
System.out.println(
"TopicTotalReceiver消费者收到消息 : " + testMessage.toString());
-
}
-
}
ok到这里我们生产者和消费者就写完了，下面我们来测试调用看看
首先启动生产者和消费者两个项目，然后postman 调用
均启动
15 Direct类型调用
消费者输出结果
16 Topic类型测试 调用message1 和message2
17 Fanout类型测试
ok 到此三种类型交换机简单收发消息就完成了， 消息回调和手动签收大家可以等我后面发帖，也可自行先百度解决。
转载：https://blog.csdn.net/Liaryank/article/details/106181089
查看评论