doker安装RabbitMQ以及用java连接

doker安装：

参考链接（非常详细）：

docker安装以及部署_docker bu shuminio_春风与麋鹿的博客-CSDN博客

安装好后开启doker

//启动docker服务
systemctl start docker
//停止docker服务
systemctl stop docker
//重启docker服务
systemctl restart docker
//查看docker服务状态
systemctl status docker

RabitMq安装：

1. 拉取do'ke中的rabbitmq镜像

docker pull rabbitmq:3.7.16-management

2.启动

docker run -d --hostname my-rabbit  -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin -p 5672:5672 -p 15672:15672  rabbitmq:3.7.16-management
 
 
参数解释：
RABBITMQ_DEFAULT_USER=admin（账号为admin）可以自己随意改
RABBITMQ_DEFAULT_PASS=admin（登录密码为admin）可以自己随意改
-p 5672:5672 ：rabbitmq的默认端口，映射到虚拟机5672端口，程序中，访问用
-p 15672:15672：rabbitmq的浏览器访问登录界面端口，映射到虚拟机15672端口，浏览器访问端口

3.浏览器访问

输入虚拟机ip和端口号进行访问

java链接

链接之前在防火墙中开发5672端口，否则会连接超时

#关闭防火墙
systemctl stop firewalld
#查看开启了哪些端口号
firewall-cmd --list-ports
#添加端口号
sudo firewall-cmd --permanent --zone=public --add-port=5672/tcp
#操作防火墙后要重启防火墙
firewall-cmd --reload

生产者：

public class Send {
    //Ctrl+Shift+U大写转换
    public static final String QUEUE_NAME = "hello";

    //发消息
    public static void main(String[] args) throws IOException, TimeoutException {
        //创建一个工厂
        ConnectionFactory factory = new ConnectionFactory();
        //工厂IP连接RabbitMQ的队列
        factory.setHost("192.168.29.129");
        //port可不设置，默认5672
//        factory.setPort(5672);
        //用户名
        factory.setUsername("admin");
        //密码
        factory.setPassword("admin");
        //创建连接
        Connection connection = factory.newConnection();
        //获取信道
        Channel channel = connection.createChannel();
        /**
         * 生成一个队列
         * 1.队列名称
         * 2.队列里面的消息是否持久化（磁盘）默认情况消息存储在内存中
         * 3.该队列是否只供一个消费者进行消费是否进行消息共享，false可以多个消费者消费 true:只能一个消费者消费
         * 4.是否自动删除最后一个消费者端开连接以后该队一句是否自动删除 true自动删除 false不自动删除
         * 5.其它参数
         */
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //发消息
        String message = "Hello World!";
        //初次使用
        /**
         * 发送一个消费
         * 1.发送到哪个交换机
         * 2.路由的Key值是哪个本次是队列的名称
         * 3.其它参数信息
         * 4.发送消息的消息体
         */
        channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
        System.out.println("消息发送完毕");

    }
}

启动：

页面显示消息已发送成功

消费者：

public class Recive {
    public static final String QUEEN_NAME = "hello";

    //接收消息
    public static void main(String[] args) throws Exception {
        //创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.29.129");
        //port可不设置，默认5672
        factory.setPort(5672);
        //用户名
        factory.setUsername("admin");
        //密码
        factory.setPassword("admin");
        Connection connection = factory.newConnection();
        //用连接创建信道
        Channel channel = connection.createChannel();

        /**
         * 消费者消费消息
         *1.消费哪个队列
         * 2.消费成功之后是否要自动应答 true 代表的自动应答 false 代表手动应答
         * 3.消费者未成功消费的回调
         * 4.消费者取消消费的回调
         */
        //声明: 接收消息
        DeliverCallback deliverCallback = (consumerTag,message) ->{
            System.out.println(message);
            System.out.println(new String(message.getBody()));
        };
        //取消消息时的回调
        CancelCallback cancelCallback = consumerTag ->{
            System.out.println("消费消息被中断");
        };

        channel.basicConsume(QUEEN_NAME,true,deliverCallback,cancelCallback);

    }
}

启动：消息接收成

简单压测
使用for循环创建20个线程，每个线程向队列中插入一百万条数据

@RequestMapping("/strongSend")
    public String strongSend(){
        for (int i = 0; i < 20; i++) {
            new Thread(() -> {
                for (int i1 = 0; i1 < 1000000; i1++) {
                    rabbitTemplate.convertAndSend(RabbitConfig.SIMPLE_QUEUE_NAME,
                           Thread.currentThread().getName());
                }
            }).start();
        }
        return "压测完成";
    }

启动项目进行压测(记得把消费者关掉，或者消费者另启一个项目)

调用压测接口