一、 环境准备
二、修改配置文件
cookie必须相同
三、防火墙添加端口
四、启动RabbitMQ
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 + " 发送失败！");
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和权限应用示例
* 说明：先阅读文档中的使用示例，创建号用户名和vhost，分配好权限。
* http://code.study.com/MQ/rabbitmq/rabbitmq/blob/master/java/src/main/java/com/study/rabbitmq/a133/cluster/readme.md#vhost使用示例
* 另外需要自己在管理界面创建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();
集群对外可以用个负载均衡技术
可以把请求分到不同的服务器上，但是只能将数据存放在一个地方，会出现单点故障
镜像队列的实现
同步信息实现高可用
镜像队列模式集群
name: 策略名称，如果使用已有的名称，保存后将会修改原来的信息
Apply to：策略应用到什么对象上
Pattern：策略应用到对象时，对象名称的匹配规则（正则表达式）
Priority：优先级，数值越大，优先级越高，相同优先级取最后一个
Definition：策略定义的类容，对于镜像队列的配置来说，只需要包含3个部分: ha-mode 、ha-params 和 ha-sync-mode。其中，ha-sync-mode是同步的方式，自动还是手动，默认是自动。ha-mode 和 ha-params 组合使用。组合方式如下：
可以查看rabitmq的日志文件查出错误
