常用命令
启动与关闭
rabbitmq-server 前台启动服务
rabbitmq-server -detached 后台启动服务
rabbitmqctl stop 停止服务
终止与启动应用
rabbitmq 进程还在,只是暂时停止队列服务
rabbitmqctl start_app 启动应用
rabbitmqctl stop_app 终止应用
用户管理
# 创建用户
rabbitmqctl add_user {username} {password}
# 删除用户
rabbitmqctl delete_user {username}
# 重置密码
rabbitmqctl change_password {username} {newpassword}
# 授予用户角色(tag)
rabbitmqctl set_user_tags {username} {tag}
# 设置用户允许访问的vhost
rabbitmqctl set_permissions -p / user_admin '.*' '.*' '.*'
RabbitMQ用户四种角色(Tag)
-
超级管理员(administrator)
可登陆管理控制台(启用management plugin的情况下),可查看所有的信息,并且可以对用户,策略(policy)进行操作。 -
监控者(monitoring)
登陆管理控制台(启用management plugin的情况下)同时可以查看rabbitmq节点的相关信息(进程数,内存使用情况,磁盘使用情况等等)。 -
策略制定者(policymarker)
登陆管理控制台(启用management plugin的情况下)同时可以对policy进行管理,但无法查看节点的相关信息 -
普通管理者(management)
仅可登陆管理控制台(启用management plugin的情况下),无法看到节点信息,也无法对策略进行管理。
AMQP
AMQP,即Advanced Message Queuing Protocol, 一个提供统一消息服务的应用层标准高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件涉及。基于此协议的客户端与消息中间件可传递消息,并不受客户端/中间件不同产品,不同开发语言等条件的限制。Erlang中实现有Rabbitmq等。
基本概念
- Producer:生产者,消息的提供者
- Consumer:消费者,消息的使用者
- Message:消息,程序间通信的数据
- Queue:队列,消息存放的容器,消息先进先出
- Vhost:虚拟主机,相当于MQ的 "数据库",用于存储队列
消息状态
Ready
消息已被送入队列,等待被消费
Unacked
- 消息已经被消费者认领,但还未被确认“已被消费”
- Unacked 状态下,消费者断开连接则消息回到 Ready
- 没有确认,客户没有断开连接,则一直处于Unacked
Finished
调用basicAck()方法后,表示消息已经被消费,从队列中移除。
RabbitMQ六种工作模式
- Hello World
- 工作队列模式(Work queues)
- 发布/订阅模式(Publish/Subscribe)
- 路由模式(Routing)
- 主题模式(Topics)
- RPC
下面我们测试一下rabbitmq这些工作模式的使用
创建一个maven工程,引入依赖:
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.9.0</version>
</dependency>
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
<version>2.8.6</version>
</dependency>
一些工具和常量类:
工具类:RabbitUtils
public class RabbitUtils {
private static ConnectionFactory connectionFactory = new ConnectionFactory();
static {
connectionFactory.setHost("127.0.0.1");
// 5672是RabbitMQ的默认端口号
connectionFactory.setPort(5672);
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
connectionFactory.setVirtualHost("/test");
}
public static Connection getConnection() {
Connection connection = null;
try {
connection = connectionFactory.newConnection();
} catch (Exception e) {
throw new RuntimeException(e);
}
return connection;
}
}
常量类:RabbitConstant
public class RabbitConstant {
public static final String QUEUE_HELLOWORLD = "helloworld";
public static final String QUEUE_SMS = "sms";
public static final String EXCHANGE_WEATHER = "weather";
public static final String EXCHANGE_WEATHER_ROUTING = "weather_routing";
public static final String QUEUE_BAIDU = "baidu";
public static final String QUEUE_SINA = "sina";
public static final String EXCHANGE_WEATHER_TOPIC = "weather_topic";
}
Hello World
先创建一个消费者进程:
public class Consumer {
public static void main(String[] args) throws IOException {
Connection conn = RabbitUtils.getConnection();
//创建通道
Channel channel = conn.createChannel();
channel.queueDeclare(RabbitConstant.QUEUE_HELLOWORLD, false, false, false, null);
//创建一个消息消费者
//第二个参数代表是否自动确认收到消息,false代表手动编程来确认消息,这是MQ的推荐做法
//第三个参数要传入DefaultConsumer的实现类
channel.basicConsume(RabbitConstant.QUEUE_HELLOWORLD, false, new MyConsumer(channel));
}
}
class MyConsumer extends DefaultConsumer {
private Channel channel;
// 重写构造函数,Channel通道对象需要从外层传入,在handleDelivery中要用到
public MyConsumer(Channel channel) {
super(channel);
this.channel = channel;
}
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
/*super.handleDelivery(consumerTag,envelope,properties,body);*/
String messageBody = new String(body);
System.out.println("消费者接收到:" + messageBody);
//签收消息,确认消息
//envelope.getDeliveryTag() 获取这个消息的TagId
//false只确认签收当前的消息,设置为true的时候则代表签收该消费者所有未签收的消息
channel.basicAck(envelope.getDeliveryTag(), false);
}
}
目前队列(QUEUE_HELLOWORLD常量的值)中还没有消息
接着再创建一个生产者进程:
public class Producer {
public static void main(String[] args) throws IOException, TimeoutException {
//TCP 物理连接
Connection conn = RabbitUtils.getConnection();
//创建通信“通道”,相当于TCP中的虚拟连接
Channel channel = conn.createChannel();
//创建队列,声明并创建一个队列,如果队列已存在,则使用这个队列
//第一个参数:队列名称ID
//第二个参数:是否持久化,false对应不持久化数据,MQ停掉数据就会丢失
//第三个参数:是否队列私有化,false则代表所有消费者都可以访问,true代表只有第一次拥有它的消费者才能一直使用,其他消费者不让访问
//第四个:是否自动删除,false代表连接停掉后不自动删除掉这个队列
//其他额外的参数, null
channel.queueDeclare(RabbitConstant.QUEUE_HELLOWORLD, false, false, false, null);
//四个参数
//exchange 交换机,暂时用不到,在后面进行发布订阅时才会用到
//队列名称
//额外的设置属性
//最后一个参数是要传递的消息字节数组
String message = "HelloWorld";
channel.basicPublish("", RabbitConstant.QUEUE_HELLOWORLD, null, message.getBytes());
channel.close();
conn.close();
System.out.println("发送数据成功");
}
}
接着查看消费者控制台的结果:
工作队列模式
- 工作队列(Work queue),它会发送一些耗时的任务给你多个工作者(Worker)。
- 在多个消息的情况下,Work Queue会将消息分派给不同的消费者,每个消费者都会收到不同的消息,并且可以根据处理消息的速度来接受消息的数量,进而让消费者程序发挥最大的性能。
- Work Queue 提别适合在集群环境中做异步处理,能最大程序发挥每一台服务器的性能。
下面以一个订票系统的简单例子来演示工作队列模式:
短信类:
public class Message {
private String name;
private String mobile;
private String content;
public Message(String name, String mobile, String content) {
this.name = name;
this.mobile = mobile;
this.content = content;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String getMobile() {
return mobile;
}
public void setMobile(String mobile) {
this.mobile = mobile;
}
public String getContent() {
return content;
}
public void setContent(String content) {
this.content = content;
}
}
三个消费者:
消费者1
public class MessageSender1 {
public static void main(String[] args) throws IOException {
Connection connection = RabbitUtils.getConnection();
final Channel channel = connection.createChannel();
channel.queueDeclare(RabbitConstant.QUEUE_SMS, false, false, false, null);
//如果不写basicQos(1),则自动MQ会将所有请求平均发送给所有消费者
//basicQos,MQ不再对消费者一次发送多个请求,而是消费者处理完一个消息后(确认后),在从队列中获取一个新的
channel.basicQos(1);//处理完一个取一个
channel.basicConsume(RabbitConstant.QUEUE_SMS , false , new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String jsonSMS = new String(body);
System.out.println("SMSSender1-短信发送成功:" + jsonSMS);
try {
Thread.sleep(10);
} catch (InterruptedException e) {
e.printStackTrace();
}
channel.basicAck(envelope.getDeliveryTag() , false);
}
});
}
}
消费2
public class MessageSender2 {
public static void main(String[] args) throws IOException {
Connection connection = RabbitUtils.getConnection();
final Channel channel = connection.createChannel();
channel.queueDeclare(RabbitConstant.QUEUE_SMS, false, false, false, null);
channel.basicQos(1);
channel.basicConsume(RabbitConstant.QUEUE_SMS , false , new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String jsonSMS = new String(body);
System.out.println("SMSSender2-短信发送成功:" + jsonSMS);
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
channel.basicAck(envelope.getDeliveryTag() , false);
}
});
}
}
消费者3
public class MessageSender3 {
public static void main(String[] args) throws IOException {
Connection connection = RabbitUtils.getConnection();
final Channel channel = connection.createChannel();
channel.queueDeclare(RabbitConstant.QUEUE_SMS, false, false, false, null);
channel.basicQos(1);
channel.basicConsume(RabbitConstant.QUEUE_SMS , false , new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String jsonSMS = new String(body);
System.out.println("SMSSender3-短信发送成功:" + jsonSMS);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
channel.basicAck(envelope.getDeliveryTag() , false);
}
});
}
}
生产者:一个订票系统
public class OrderSystem {
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = RabbitUtils.getConnection();
Channel channel = connection.createChannel();
channel.queueDeclare(RabbitConstant.QUEUE_SMS, false, false, false, null);
for(int i = 100 ; i <= 200 ; i++) {
Message message = new Message("乘客" + i, "13900000" + i, "您的车票已预订成功");
String messageJsonStr = new Gson().toJson(message);
channel.basicPublish("" , RabbitConstant.QUEUE_SMS , null , messageJsonStr.getBytes());
}
System.out.println("发送数据成功");
channel.close();
connection.close();
}
}
生产者发送101条消息到RabbitConstant.QUEUE_SMS
队列:
三个消费者一共消费这101条消息:
发布(Publish)/订阅(Subscribe)模式
- 发布/订阅模式中,生产者不再直接与队列绑定,而是将数据发送至 交换机
Exchange
- 交换机用于将数据按某种规则送入与之绑定的队列,进而供消费者使用。
- 发布/订阅模式中,交换机将无差别的将所有消息送入与之绑定的队列,所有消费者拿到的消息完全相同,交换机的类型为
fanout
。
适用于数据提供商与应用商,例如,气象局提供天气数据送入交换机,各大平台(百度,新浪等)接入通过队列绑定到该交换机,自动获取气象局推送的气象数据。
下面就通过代码的方式来实现一个天气的例子
使用本机的rabbitmq(单机模式),在rabbitmq管理页面手动创建交换器和队列
队列baidu和sina会自动创建
大平台-百度(消费者)
public class Baidu {
public static void main(String[] args) throws IOException {
Connection connection = RabbitUtils.getConnection();
final Channel channel = connection.createChannel();
channel.queueDeclare(RabbitConstant.QUEUE_BAIDU, false, false, false, null);
//queueBind用于将队列与交换机绑定
//参数1:队列名 参数2:交互机名 参数三:路由key(暂时用不到)
channel.queueBind(RabbitConstant.QUEUE_BAIDU, RabbitConstant.EXCHANGE_WEATHER, "");
channel.basicQos(1);
channel.basicConsume(RabbitConstant.QUEUE_BAIDU , false , new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("百度收到气象信息:" + new String(body));
channel.basicAck(envelope.getDeliveryTag() , false);
}
});
}
}
大平台-新浪(消费者)
public class Sina {
public static void main(String[] args) throws IOException {
Connection connection = RabbitUtils.getConnection();
final Channel channel = connection.createChannel();
channel.queueDeclare(RabbitConstant.QUEUE_SINA, false, false, false, null);
//queueBind用于将队列与交换机绑定
//参数1:队列名 参数2:交互机名 参数三:路由key(暂时用不到)
channel.queueBind(RabbitConstant.QUEUE_SINA, RabbitConstant.EXCHANGE_WEATHER, "");
channel.basicQos(1);
channel.basicConsume(RabbitConstant.QUEUE_SINA , false , new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("新浪收到气象信息:" + new String(body));
channel.basicAck(envelope.getDeliveryTag() , false);
}
});
}
}
气象台(生产者)
public class WeatherBureau {
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = RabbitUtils.getConnection();
String message = "上海明天天气,晴转多云!";
Channel channel = connection.createChannel();
channel.basicPublish(RabbitConstant.EXCHANGE_WEATHER, "", null, message.getBytes());
channel.close();
connection.close();
System.out.println("天气发布成功");
}
}
执行结果:
路由Routinig模式
- 路由模式是在发布订阅模式基础上的变种。
- 发布订阅是无条件将所有消息分发给所有消费者队列。
- 路由模式则是交换机根据
Routing Key
有条件的将数据筛选后发给消费者队列。 - 路由模式下交换机的类型被称为
direct
。
下面还是通过一个例子来测试下:
手动创建一个带路由件的交换器:
大平台-百度(消费者)
public class Baidu {
public static void main(String[] args) throws IOException {
Connection connection = RabbitUtils.getConnection();
final Channel channel = connection.createChannel();
channel.queueDeclare(RabbitConstant.QUEUE_BAIDU, false, false, false, null);
//queueBind用于将队列与交换机绑定
//参数1:队列名 参数2:交互机名 参数三:路由key
channel.queueBind(RabbitConstant.QUEUE_BAIDU, RabbitConstant.EXCHANGE_WEATHER_ROUTING, "china.jiangsu.xuzhou.20210311");
channel.queueBind(RabbitConstant.QUEUE_BAIDU, RabbitConstant.EXCHANGE_WEATHER_ROUTING, "china.jiangsu.shuqian.20210311");
channel.queueBind(RabbitConstant.QUEUE_BAIDU, RabbitConstant.EXCHANGE_WEATHER_ROUTING, "china.jiangsu.lianyungang.20210311");
channel.queueBind(RabbitConstant.QUEUE_BAIDU, RabbitConstant.EXCHANGE_WEATHER_ROUTING, "china.jiangsu.huaian.20210311");
channel.queueBind(RabbitConstant.QUEUE_BAIDU, RabbitConstant.EXCHANGE_WEATHER_ROUTING, "china.jiangsu.yancheng.20210311");
channel.basicQos(1);
channel.basicConsume(RabbitConstant.QUEUE_BAIDU , false , new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("百度收到气象信息:" + new String(body));
channel.basicAck(envelope.getDeliveryTag() , false);
}
});
}
}
大平台-新浪(消费者)
public class Sina {
public static void main(String[] args) throws IOException {
Connection connection = RabbitUtils.getConnection();
final Channel channel = connection.createChannel();
channel.queueDeclare(RabbitConstant.QUEUE_SINA, false, false, false, null);
channel.queueBind(RabbitConstant.QUEUE_SINA, RabbitConstant.EXCHANGE_WEATHER_ROUTING, "china.jiangsu.nanjing.20210311");
channel.queueBind(RabbitConstant.QUEUE_SINA, RabbitConstant.EXCHANGE_WEATHER_ROUTING, "china.jiangsu.yangzhou.20210311");
channel.queueBind(RabbitConstant.QUEUE_SINA, RabbitConstant.EXCHANGE_WEATHER_ROUTING, "china.jiangsu.taizhou.20210311");
channel.queueBind(RabbitConstant.QUEUE_SINA, RabbitConstant.EXCHANGE_WEATHER_ROUTING, "china.jiangsu.nantong.20210311");
channel.queueBind(RabbitConstant.QUEUE_SINA, RabbitConstant.EXCHANGE_WEATHER_ROUTING, "china.jiangsu.zhenjiang.20210311");
channel.basicQos(1);
channel.basicConsume(RabbitConstant.QUEUE_SINA , false , new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("新浪收到气象信息:" + new String(body));
channel.basicAck(envelope.getDeliveryTag() , false);
}
});
}
}
气象台(生产者)
public class WeatherBureau {
public static void main(String[] args) throws IOException, TimeoutException {
Map<String, String> area = new LinkedHashMap<>();
// 苏北五虎
area.put("china.jiangsu.xuzhou.20210311", "中国江苏<<徐州>>2021年3月11日天气数据");
area.put("china.jiangsu.shuqian.20210311", "中国江苏<<宿迁>>2021年3月11日天气数据");
area.put("china.jiangsu.lianyungang.20210311", "中国江苏<<连云港>>2021年3月11日天气数据");
area.put("china.jiangsu.huaian.20210311", "中国江苏<<淮安>>2021年3月11日天气数据");
area.put("china.jiangsu.yancheng.20210311", "中国江苏<<盐城>>2021年3月11日天气数据");
// 苏中
area.put("china.jiangsu.nanjing.20210311", "中国江苏<<南京>>2021年3月11日天气数据");
area.put("china.jiangsu.yangzhou.20210311", "中国江苏<<扬州>>2021年3月11日天气数据");
area.put("china.jiangsu.taizhou.20210311", "中国江苏<<泰州>>2021年3月11日天气数据");
area.put("china.jiangsu.nantong.20210311", "中国江苏<<南通>>2021年3月11日天气数据");
area.put("china.jiangsu.zhenjiang.20210311", "中国江苏<<镇江>>2021年3月11日天气数据");
Connection connection = RabbitUtils.getConnection();
Channel channel = connection.createChannel();
for (Map.Entry<String, String> me : area.entrySet()) {
//Routing key 第二个参数相当于数据筛选的条件
channel.basicPublish(RabbitConstant.EXCHANGE_WEATHER_ROUTING, me.getKey(), null, me.getValue().getBytes());
}
channel.close();
connection.close();
}
}
运行代码,先启动消费者,再启动生产者,可以清楚看到消费者根据路由件去匹配获取的数据,百度消费的是苏北的天气情况,新浪消费的是苏中的天气情况。
主题Topic模式
- 主题模式是在Routing模式基础上,提供了对Route Key模式匹配的功能,可以简化程序的编写。
- 主题模式下,模糊匹配表达式规则为
* 匹配单个关键字
# 匹配所有关键字
- 主题模式下交换机的类型为
topic
手动新建一个主题模式的交换器:
生产者:
public class WeatherBureau {
public static void main(String[] args) throws IOException, TimeoutException {
Map<String, String> area = new LinkedHashMap<>();
// 苏北
area.put("china.jiangsu.xuzhou.20210311", "中国江苏<<徐州>>2021年3月11日天气数据");
area.put("china.jiangsu.shuqian.20210311", "中国江苏<<宿迁>>2021年3月11日天气数据");
area.put("china.jiangsu.lianyungang.20210311", "中国江苏<<连云港>>2021年3月11日天气数据");
area.put("china.jiangsu.huaian.20210311", "中国江苏<<淮安>>2021年3月11日天气数据");
area.put("china.jiangsu.yancheng.20210311", "中国江苏<<盐城>>2021年3月11日天气数据");
// 上海·
area.put("china.shanghai.songjiangqu.20210311", "中国上海<<松江区>>2021年3月11日天气数据");
area.put("china.shanghai.jinganqu.20210312", "中国上海<<静安区>>2021年3月12日天气数据");
area.put("china.shanghai.xuhuiqu.20210312", "中国上海<<徐汇区>>2021年3月12日天气数据");
Connection connection = RabbitUtils.getConnection();
Channel channel = connection.createChannel();
Iterator<Map.Entry<String, String>> itr = area.entrySet().iterator();
while (itr.hasNext()) {
Map.Entry<String, String> me = itr.next();
//Routing key 第二个参数相当于数据筛选的条件
channel.basicPublish(RabbitConstant.EXCHANGE_WEATHER_TOPIC,me.getKey() , null , me.getValue().getBytes());
}
channel.close();
connection.close();
System.out.println("天气数据发布成功");
}
}
消费者:
public class Baidu {
public static void main(String[] args) throws IOException {
Connection connection = RabbitUtils.getConnection();
final Channel channel = connection.createChannel();
channel.queueDeclare(RabbitConstant.QUEUE_BAIDU, false, false, false, null);
//queueBind用于将队列与交换机绑定
//参数1:队列名 参数2:交互机名 参数三:路由key
channel.queueBind(RabbitConstant.QUEUE_BAIDU, RabbitConstant.EXCHANGE_WEATHER_TOPIC, "china.jiangsu.#");
channel.basicQos(1);
channel.basicConsume(RabbitConstant.QUEUE_BAIDU , false , new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("百度收到气象信息:" + new String(body));
channel.basicAck(envelope.getDeliveryTag() , false);
}
});
}
}
public class Sina {
public static void main(String[] args) throws IOException {
Connection connection = RabbitUtils.getConnection();
final Channel channel = connection.createChannel();
channel.queueDeclare(RabbitConstant.QUEUE_SINA, false, false, false, null);
channel.queueBind(RabbitConstant.QUEUE_SINA, RabbitConstant.EXCHANGE_WEATHER_TOPIC, "*.*.*.20210312");
channel.basicQos(1);
channel.basicConsume(RabbitConstant.QUEUE_SINA, false, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("新浪收到气象信息:" + new String(body));
channel.basicAck(envelope.getDeliveryTag(), false);
}
});
}
}
结果:
RabbitMQ消息确认机制
- RabbitMQ在传递消息的过程中充当了代理人(Broker)的角色,那生产者(Producer)怎样知道消息被正确投递到Broker了呢?
- RabbitMQ提供了监听器(Listener)来接收消息投递的状态。
- 消息确认涉及两种状态:Confirm和Return。
Confirm和Return
- Confirm 代表生产者将消息送到Broker时产生的状态,后续会出现两种情况:
ack 代表Broker已经将数据接收
nack 代表Broker拒收消息。原因可能是:队列已满,限流,IO异常等等 - Return 代表消息被Broker正常接收(ack)后,但Broker没有对应的队列进行投递时产生的状态,消息被退给生产者。
- 注意:上面两种状态只代表生产者与Broker之间消息投递的情况。与消费者是否接收/确认消息无关。
消费者:
public class Baidu {
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = RabbitUtils.getConnection();
final Channel channel = connection.createChannel();
channel.queueDeclare(RabbitConstant.QUEUE_BAIDU, false, false, false, null);
//queueBind用于将队列与交换机绑定
//参数1:队列名 参数2:交互机名 参数三:路由key
channel.queueBind(RabbitConstant.QUEUE_BAIDU, RabbitConstant.EXCHANGE_WEATHER_TOPIC, "china.jiangsu.#");
channel.basicQos(1);
channel.basicConsume(RabbitConstant.QUEUE_BAIDU, false, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("百度收到气象信息:" + new String(body) + ", DeliveryTag: " + envelope.getDeliveryTag());
if ("china.jiangsu.lianyungang.20210311".equals(envelope.getRoutingKey())) {
channel.basicNack(envelope.getDeliveryTag(), false, true);
} else {
channel.basicAck(envelope.getDeliveryTag(), false);
}
}
});
}
}
在百度队列中有五条消息,当消费到第三条时,我做了一个不确认操作,把此消息再重新放回队列,所以就看到下面这种结果(只成功消费了两条):
这第三条一直在返回队列,取出来消费做无限循环操作,导致后面两条消息也没机会消费。
如果把上面的代码:channel.basicNack(envelope.getDeliveryTag(), false, true);
改成channel.basicNack(envelope.getDeliveryTag(), false, false);
第三个参数改成了false,那么此消息就相当于被丢弃了,不会重新放入队列。
太平台-新浪(消费者)
public class Sina {
public static void main(String[] args) throws IOException {
Connection connection = RabbitUtils.getConnection();
final Channel channel = connection.createChannel();
channel.queueDeclare(RabbitConstant.QUEUE_SINA, false, false, false, null);
channel.queueBind(RabbitConstant.QUEUE_SINA, RabbitConstant.EXCHANGE_WEATHER_TOPIC, "*.*.*.20210312");
channel.basicQos(1);
channel.basicConsume(RabbitConstant.QUEUE_SINA, false, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("新浪收到气象信息:" + new String(body) + ", DeliveryTag: " + envelope.getDeliveryTag());
channel.basicAck(envelope.getDeliveryTag(), false);
}
});
}
}
生产者
public class WeatherBureau {
public static void main(String[] args) throws IOException, TimeoutException {
Map<String, String> area = new LinkedHashMap<>();
// 苏北
area.put("china.jiangsu.xuzhou.20210311", "中国江苏<<徐州>>2021年3月11日天气数据");
area.put("china.jiangsu.shuqian.20210311", "中国江苏<<宿迁>>2021年3月11日天气数据");
area.put("china.jiangsu.lianyungang.20210311", "中国江苏<<连云港>>2021年3月11日天气数据");
area.put("china.jiangsu.huaian.20210311", "中国江苏<<淮安>>2021年3月11日天气数据");
area.put("china.jiangsu.yancheng.20210311", "中国江苏<<盐城>>2021年3月11日天气数据");
// 上海·
area.put("china.shanghai.songjiangqu.20210311", "中国上海<<松江区>>2021年3月11日天气数据");
area.put("china.shanghai.jinganqu.20210312", "中国上海<<静安区>>2021年3月12日天气数据");
area.put("china.shanghai.xuhuiqu.20210312", "中国上海<<徐汇区>>2021年3月12日天气数据");
Connection connection = RabbitUtils.getConnection();
Channel channel = connection.createChannel();
//开启confirm监听模式
channel.confirmSelect();
channel.addConfirmListener(new ConfirmListener() {
@Override
public void handleAck(long deliveryTag, boolean multiple) throws IOException {
//第二个参数代表接收的数据是否为批量接收,一般我们用不到。
System.out.println("消息已被Broker 接收, deliveryTag: " + deliveryTag + ", multiple: " + multiple);
}
@Override
public void handleNack(long deliveryTag, boolean multiple) throws IOException {
System.out.println("消息已被Broker 拒收, deliveryTag: " + deliveryTag + ", multiple: " + multiple);
}
});
channel.addReturnListener(new ReturnCallback() {
@Override
public void handle(Return r) {
System.err.println("===========================");
System.err.println("Return编码:" + r.getReplyCode() + ", Return描述:" + r.getReplyText());
System.err.println("交换器:" + r.getExchange() + ", 路由key:" + r.getRoutingKey());
System.err.println("Return主题:" + new String(r.getBody()));
System.err.println("===========================");
}
});
for (Map.Entry<String, String> me : area.entrySet()) {
//Routing key 第二个参数相当于数据筛选的条件
//第三个参数为:mandatory true代表如果消息无法正常投递则return回生产者,如果false,则直接将消息放弃。
channel.basicPublish(RabbitConstant.EXCHANGE_WEATHER_TOPIC, me.getKey(), true, null, me.getValue().getBytes());
}
/*channel.close();
connection.close();*/
}
}
在生产者中开启了Confirm和Return 的监听,看下生产者的执行结果
可以看到上海松江区那条消息找不到队列给退回生产者了。
RabbitMQ集群架构模式
四种架构模式:
- 主备模式(Warren)
- 镜像模式(Mirror)
- 远程模式(Shovel)
- 多活模式(Federation)
主备模式(Warren)
实现RabbitMQ的高可用集群,一般在并发和数据量不高的情况下,这种模型非常好用且简单。
镜像模式(Mirror)
集群模式非常经典的就是镜像模式,保证100%数据不丢失,在实际工作中是使用最多的。并且实现集群非常的简单,一般大公司都会使用这种镜像模式。
远程模式(Shovel)
远程模式可以实现双活的一种模式,简称Shovel模式,所谓Shovel就是我们可以把消息进行不同数据中心的复制工作,可以跨地域的让两个mq集群互联。
多活模式(Federation)
多活模式是实现异地数据复制的主流模式,因为Shovel模式配置比较复杂,所以一般来说实现异地集群都是使用这种双活或多活模型来实现的。这种模型需要依赖rabbitmq的federation插件,可以实现持续的可靠的AMQP数据通信,多活模式在实际配置与应用非常的简单。
评论