1.配置rabbitmq集群（可以参考前一篇RabbitMq之部署集群）

2.Nginx做负载均衡

注意:Nginx1.90版本后 新增了stream 模块用于一般的 TCP 代理和负载均衡，之前版本不支持

修改Nginx配置文件nginx.conf添加如下配置，监听12345端口

stream { 
    upstream rabbitmqserver { 
        server 192.168.191.20:5672; 
        server 192.168.191.131:5672;
        server 192.168.191.132:5672;
    }

server { 
        listen 12345; 
        proxy_pass rabbitmqserver; 
    } 
 
}

修改后重启Nginx，使之生效。

3.测试

获取连接的工具类

package com.sky.study;

import java.io.IOException;

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/**
 * 生成连接
 * @author 86940
 *
 */
public class ConnectionUtil {
    public static Connection getConnection() throws IOException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.191.132");//做Nginx负载均衡的服务器地址
        factory.setPort(12345);//使用Nginx做了tcp负载均衡，监听的是12345端口
        factory.setUsername("admin");
        factory.setPassword("123456");
        factory.setVirtualHost("hello");
        Connection connection = factory.newConnection();
        return connection;
    }
}

生产者

package com.sky.study.helloWorld;

import java.io.IOException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.sky.study.ConnectionUtil;

public class Send {
    private static final String QUEUE_NAME = "q.test.01";

public static void main(String[] args) {
        Connection connection = null;
        Channel channel = null;
        try {
            // 获取连接
            connection = ConnectionUtil.getConnection();
            // 创建通过
            channel = connection.createChannel();
            // 声明（创建）队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            // 消息
            String message = "Hello World!";
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println("发送成功");

} 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();
                }
            }
        }

}
}
消费者

package com.sky.study.helloWorld;

import java.io.IOException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;
import com.sky.study.ConnectionUtil;
import com.rabbitmq.client.ShutdownSignalException;

public class Recv {
    private static final String QUEUE_NAME = "q.test.01";

public static void main(String[] args) {
        Connection connection = null;
        Channel channel = null;
        try {
            connection = ConnectionUtil.getConnection();
            channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            // 定义队列的消费者
            QueueingConsumer consumer = new QueueingConsumer(channel);
            // 监听队列
            channel.basicConsume(QUEUE_NAME, true, consumer);
            // 获取消息
            while (true) {
                Delivery nextDelivery = consumer.nextDelivery();
                String message = new String(nextDelivery.getBody());
                System.out.println(message);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (ShutdownSignalException e) {
            e.printStackTrace();
        } catch (ConsumerCancelledException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

结束，谢谢支持