RabbitMQ

100 阅读 0 评论 66 点赞

我是靠谱客的博主寒冷玉米，最近开发中收集的这篇文章主要介绍RabbitMQ，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

1.在linux中使用docker-compose.yml安装

建立文件后输入：

version: "3.1"
services:
  rabbitmq:
    image: 10.0.134.175:5000/rabbitmq:3.7.14
    restart: always
    container_name: rabbitmq
    ports:
      - 5672:5672
      - 15672:15672
    volumes:
      - ./data:/var/lib/rabbitmq

2.运行该文件

3.在浏览器中输入地址，打开，输入账号密码

4.先写一个连接mq的工具类

package com.qianfeng.rabbitmq.utils;

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @Author kk
 * @Date 2020/7/10 19:45
 */
public class ConnectionUtil {

    /**
     * 获取连接的方法
     * @return
     * @throws Exception
     */
    public static Connection getConnection() throws Exception{
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.60.128"); //Linux主机地址
        connectionFactory.setPort(5672); //主机端口
        connectionFactory.setUsername("test");
        connectionFactory.setPassword("test");
        connectionFactory.setVirtualHost("/test");
        Connection connection = connectionFactory.newConnection();
        return connection;
    }
}

5. 代码展示生产者消费者

Sender(生产者代码)：

package com.qianfeng.rabbitmq.simple;

import com.qianfeng.rabbitmq.utils.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

/**
 * 这是生产者
 * @Author kk
 * @Date 2020/7/10 19:44
 */
public class Sender {

    private static final String QUEUE_NAME = "wcc";

    public static void main(String[] args) throws Exception {
        //连接mq服务器
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel(); //创建通道，类似于数据库的statement
        /**
         * 声明队列，队列要先声明才能使用，队列一般都是生产者和消费者都声明，因为你无法保证生产和消费哪个先启动
         * 防止互相找不到；如果声明队列的时候队列存在，则忽略，如果不存在，就创建；但是注意，如果队列已经存在了，
         * 再次声明的时候如果其中的参数不一致会抛出异常
         * 参数1：队列的名字
         * 参数2：是否持久化：队列默认是保存在内存中的，如果设置了持久化，会自动保存到磁盘上(mq内部有一个数据库)
         * 如果不保存，mq重启服务器后数据会丢失
         * 参数3：是否排外（有两个功能）：功能1：当消费者断开后是不是自动删除当前队列，功能2：设置当前队列是否排外，代表当前队列是不是私有的，私有的就是只允许一个消费者连接
         * 当前队列，如果有多个消费者同时连接，会抛出异常，相当于加了一个锁
         * 参数4：当最后一个消费者断开连接后是不是自动删除当前队列
         * 参数5：一些额外的参数
         */
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        String message = "当前是队列中的第";
        for (int i = 1; i <= 50; i++) {
            channel.basicPublish("",QUEUE_NAME,null,(message+i+"个消息").getBytes());
        }
        channel.close();
        connection.close();
    }
}

6.运行改代码后

7.消费者(Consumer)

package com.qianfeng.rabbitmq.simple;

import com.qianfeng.rabbitmq.utils.ConnectionUtil;
import com.rabbitmq.client.*;


import java.io.IOException;
import java.util.Date;

/**
 * 这是消费者
 * @Author kk
 * @Date 2020/7/10 20:09
 */
public class Consumer {
    //因为是sender与consumer互相发送消息，所以在的队列必须一样，才能在同一个队列中取出
    private static final String QUEUE_NAME = "wcc";

    public static void main(String[] args) throws Exception {

        //连接mq服务器
        Connection connection = ConnectionUtil.getConnection();
        Channel channel = connection.createChannel(); //创建通道，类似于数据库的statement

        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        /**
         * 消费消息
         * 参数1：队列的名字
         * 参数2：收到消息是否自动应答，就是收到消息后告诉服务器我收到了，如果该项什么都不填，则每启动一次消费者都会收到消息；填为true后，只收到一次，之后再启动该消费者，也不会收到消息
         * 参数3：如何处理消息
         */
        channel.basicConsume(QUEUE_NAME,true,new DefaultConsumer(channel){
            /**
             * 处理消息的方法，得等待消息过来，消息来了处理消息
             * @param consumerTag
             * @param envelope
             * @param properties
             * @param body
             * @throws IOException
             */
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                String message = new String(body);
                System.err.println("当前时间为："+new Date()+message);
            }
        });
         //消费者是不能关闭连接的，就像我们收短信的手机一样不能关机，因为关机之后就收不到短信了
    }
}

8.因为生产者刚才运行了一边，现在直接运行消费者代码

如下图所示: