消息中间件-RabbitMQ集群和高可用
2024-10-15 03:41:28
多机多节点集群部署
一、 环境准备
准备三台安装好RabbitMQ 的机器，安装方法见 安装步骤
- 10.10.1.41
- 10.10.1.42
- 10.10.1.43
提示：如果使用虚拟机，可以在一台VM上安装好RabbitMQ后，创建快照，从快照创建链接克隆，会节省很多磁盘空间
二、修改配置文件
cookie必须相同
三、防火墙添加端口
四、启动RabbitMQ
package com.study.rabbitmq.a133.cluster;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.TimeoutException;
/**
* 客户端连接集群示例
*/
public class Consumer {
public static void main(String[] args) {
// 1、创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
// 2、设置连接属性
factory.setUsername("order-user");
factory.setPassword("order-user");
factory.setVirtualHost("v1");
Connection connection = null;
Channel channel = null;
// 3、设置每个节点的链接地址和端口
Address[] addresses = new Address[]{
new Address("192.168.100.242", 5672),
new Address("192.168.100.241", 5672)
};
try {
// 开启/关闭连接自动恢复，默认是开启状态
factory.setAutomaticRecoveryEnabled(true);
// 设置每100毫秒尝试恢复一次，默认是5秒：com.rabbitmq.client.ConnectionFactory.DEFAULT_NETWORK_RECOVERY_INTERVAL
factory.setNetworkRecoveryInterval(100);
// 4、从连接工厂获取连接
connection = factory.newConnection(addresses, "消费者");
// 添加重连监听器
((Recoverable) connection).addRecoveryListener(new RecoveryListener() {
/**
* 重连成功后的回调
* @param recoverable
*/
public void handleRecovery(Recoverable recoverable) {
System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SS").format(new Date()) + " 已重新建立连接！");
}
/**
* 开始重连时的回调
* @param recoverable
*/
public void handleRecoveryStarted(Recoverable recoverable) {
System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SS").format(new Date()) + " 开始尝试重连！");
}
});
// 5、从链接中创建通道
channel = connection.createChannel();
/**
* 6、声明（创建）队列
* 如果队列不存在，才会创建
* RabbitMQ 不允许声明两个队列名相同，属性不同的队列，否则会报错
*
* queueDeclare参数说明：
* @param queue 队列名称
* @param durable 队列是否持久化
* @param exclusive 是否排他，即是否为私有的，如果为true,会对当前队列加锁，其它通道不能访问，
* 并且在连接关闭时会自动删除，不受持久化和自动删除的属性控制。
* 一般在队列和交换器绑定时使用
* @param autoDelete 是否自动删除，当最后一个消费者断开连接之后是否自动删除
* @param arguments 队列参数，设置队列的有效期、消息最大长度、队列中所有消息的生命周期等等
*/
channel.queueDeclare("queue1", true, false, false, null);
// 7、定义收到消息后的回调
final Channel finalChannel = channel;
DeliverCallback callback = new DeliverCallback() {
public void handle(String consumerTag, Delivery message) throws IOException {
System.out.println("收到消息：" + new String(message.getBody(), "UTF-8"));
finalChannel.basicAck(message.getEnvelope().getDeliveryTag(), false);
}
};
// 8、监听队列
channel.basicConsume("queue1", false, callback, new CancelCallback() {
public void handle(String consumerTag) throws IOException {
}
});
System.out.println("开始接收消息");
System.in.read();
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
} finally {
// 9、关闭通道
if (channel != null && channel.isOpen()) {
try {
channel.close();
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
}
// 10、关闭连接
if (connection != null && connection.isOpen()) {
try {
connection.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
}
package com.study.rabbitmq.a133.cluster;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
/**
* 客户端连接集群示例
*/
public class Producer {
public static void main(String[] args) {
// 1、创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
// 2、设置连接属性
factory.setUsername("order-user");
factory.setPassword("order-user");
factory.setVirtualHost("v1");
Connection connection = null;
Channel channel = null;
// 3、设置每个节点的链接地址和端口
Address[] addresses = new Address[]{
new Address("192.168.100.242", 5672),
new Address("192.168.100.241", 5672)
};
try {
// 开启/关闭连接自动恢复，默认是开启状态
factory.setAutomaticRecoveryEnabled(true);
// 设置每100毫秒尝试恢复一次，默认是5秒：com.rabbitmq.client.ConnectionFactory.DEFAULT_NETWORK_RECOVERY_INTERVAL
factory.setNetworkRecoveryInterval(100);
factory.setTopologyRecoveryEnabled(false);
// 4、使用连接集合里面的地址获取连接
connection = factory.newConnection(addresses, "生产者");
// 添加重连监听器
((Recoverable) connection).addRecoveryListener(new RecoveryListener() {
/**
* 重连成功后的回调
* @param recoverable
*/
public void handleRecovery(Recoverable recoverable) {
System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SS").format(new Date()) + " 已重新建立连接！");
}
/**
* 开始重连时的回调
* @param recoverable
*/
public void handleRecoveryStarted(Recoverable recoverable) {
System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SS").format(new Date()) + " 开始尝试重连！");
}
});
// 5、从链接中创建通道
channel = connection.createChannel();
/**
* 6、声明（创建）队列
* 如果队列不存在，才会创建
* RabbitMQ 不允许声明两个队列名相同，属性不同的队列，否则会报错
*
* queueDeclare参数说明：
* @param queue 队列名称
* @param durable 队列是否持久化
* @param exclusive 是否排他，即是否为私有的，如果为true,会对当前队列加锁，其它通道不能访问，并且在连接关闭时会自动删除，不受持久化和自动删除的属性控制
* @param autoDelete 是否自动删除，当最后一个消费者断开连接之后是否自动删除
* @param arguments 队列参数，设置队列的有效期、消息最大长度、队列中所有消息的生命周期等等
*/
channel.queueDeclare("queue1", true, false, false, null);
for (int i = 0; i < 100; i++) {
// 消息内容
String message = "Hello World " + i;
try {
// 7、发送消息
channel.basicPublish("", "queue1", null, message.getBytes());
} catch (AlreadyClosedException e) {
// 可能连接已关闭，等待重连
System.out.println("消息 " + message + " 发送失败！");
i--;
TimeUnit.SECONDS.sleep(2);
continue;
}
System.out.println("消息 " + i + " 已发送！");
TimeUnit.SECONDS.sleep(2);
}
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
// 8、关闭通道
if (channel != null && channel.isOpen()) {
try {
channel.close();
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
}
// 9、关闭连接
if (connection != null && connection.isOpen()) {
try {
connection.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
}
package com.study.rabbitmq.a133.cluster;
import com.rabbitmq.client.*;
import com.rabbitmq.client.Consumer;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* vhost和权限应用示例
* <p>
* 说明：先阅读文档中的使用示例，创建号用户名和vhost，分配好权限。
* http://code.study.com/MQ/rabbitmq/rabbitmq/blob/master/java/src/main/java/com/study/rabbitmq/a133/cluster/readme.md#vhost使用示例
* <p>
* 另外需要自己在管理界面创建queue2队列和test交换器
*/
public class VirtualHostsDemo {
public static void main(String[] args) {
// 1、创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
// 2、设置连接属性
factory.setUsername("order-user");
factory.setPassword("order-user");
factory.setVirtualHost("v1");
Connection connection = null;
Channel prducerChannel = null;
Channel consumerChannel = null;
// 3、设置每个节点的链接地址和端口
Address[] addresses = new Address[]{
new Address("192.168.100.242", 5672),
new Address("192.168.100.241", 5672)
};
try {
// 4、从连接工厂获取连接
connection = factory.newConnection(addresses, "消费者");
// 5、从链接中创建通道
prducerChannel = connection.createChannel();
prducerChannel.exchangeDeclare("test-exchange", "fanout");
prducerChannel.queueDeclare("queue1", false, false, false, null);
prducerChannel.queueBind("queue1", "test-exchange", "xxoo");
// 消息内容
String message = "Hello A";
prducerChannel.basicPublish("test-exchange", "c1", null, message.getBytes());
consumerChannel = connection.createChannel();
// 创建一个消费者对象
Consumer consumer = new DefaultConsumer(consumerChannel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("收到消息：" + new String(body, "UTF-8"));
}
};
consumerChannel.basicConsume("queue1", true, consumer);
System.out.println("等待接收消息");
System.in.read();
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
} finally {
// 9、关闭通道
if (prducerChannel != null && prducerChannel.isOpen()) {
try {
prducerChannel.close();
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
}
// 10、关闭连接
if (connection != null && connection.isOpen()) {
try {
connection.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
}
集群对外可以用个负载均衡技术
可以把请求分到不同的服务器上，但是只能将数据存放在一个地方，会出现单点故障
镜像队列的实现
同步信息实现高可用
镜像队列模式集群
- 镜像队列属于RabbitMQ 的高可用方案，见：https://www.rabbitmq.com/ha.html#mirroring-arguments
- 通过前面的步骤搭建的集群属于普通模式集群，是通过共享元数据实现集群
开启镜像队列模式需要在管理页面添加策略，添加方式：
- 进入管理页面 -> Admin -> Policies（在页面右侧）-> Add / update a policy
在表单中填入：
参数说明
name: 策略名称，如果使用已有的名称，保存后将会修改原来的信息
Apply to：策略应用到什么对象上
Pattern：策略应用到对象时，对象名称的匹配规则（正则表达式）
Priority：优先级，数值越大，优先级越高，相同优先级取最后一个
Definition：策略定义的类容，对于镜像队列的配置来说，只需要包含3个部分: ha-mode 、ha-params 和 ha-sync-mode。其中，ha-sync-mode是同步的方式，自动还是手动，默认是自动。ha-mode 和 ha-params 组合使用。组合方式如下：
镜像队列模式相比较普通模式，镜像模式会占用更多的带宽来进行同步，所以镜像队列的吞吐量会低于普通模式
- 但普通模式不能实现高可用，某个节点挂了后，这个节点上的消息将无法被消费，需要等待节点启动后才能被消费。
可以查看rabitmq的日志文件查出错误
常用命令
RabbitMQ高可用集群方案
单点失效会存在
由插件负责同步，但会出现数据不一致
最新文章
- JSONP的诞生、原理及应用实例
- MFC 如何创建浏览文件夹的对话框
- springMVC配置Json
- RPC和Socket,RMI和RPC之间的关系
- sharding-jdbc-how2work 当当的sharding-jdbc剖析(查询)
- JQuery识别键盘操作 & 键盘快捷键
- Servlet的延迟加载和预加载
- Codeforces Beta Round #12 (Div 2 Only) D. Ball sort/map
- Android实例-读取设备联系人(XE8+小米2)
- bind()
- 编程的宗派（OOP与FP孰优孰劣）--王垠
- [原]Sublime Text3 搭建16位汇编环境（windows）
- 移动收入超PC端 盛大文学战略转型初见成效
- [Swift]LeetCode82. 删除排序链表中的重复元素 II | Remove Duplicates from Sorted List II
- Pandas系列（四）-文本数据处理
- 痞子衡嵌入式：SEGGER J-Link仿真器硬件版本变迁
- [LeetCode]题100：Same Tree
- HTML 标记 1
- numpy 初识（二）
- P2865 [USACO06NOV]路障Roadblocks