一篇文章带你从入门到精通:RabbitMQ

[root@centos7 rabbitmq]# ls
esl-erlang_23.2.3-1_centos_7_amd64.rpm  rabbitmq-server-3.8.14-1.el7.noarch.rpm
[root@centos7 rabbitmq]# pwd
/usr/local/bin/rabbitmq

# 安装 erlang，安装后执行 erl -v 显示版本号则代表成功
rpm -ivh esl-erlang_23.2.3-1_centos_7_amd64.rpm
# 安装 socat 这里没有下载源文件，而是直接通过 yum 在线安装，因为它并不大
yum install -y socat
# 安装 rabbitmq
rpm -ivh rabbitmq-server-3.8.14-1.el7.noarch.rpm

# 启动服务
systemctl start rabbitmq-server
# 开机自启
systemctl enable rabbitmq-server
# 停止服务
systemctl stop rabbitmq-server
# 查看服务状态
systemctl status rabbitmq-server.service

# 安装命令
rabbitmq-plugins enable rabbitmq_management
# 重启服务
systemctl restart rabbitmq-server

# 查询 15672 是否开放，一般默认都是 no
firewall-cmd --query-port=15672/tcp
# 开放指定端口 15672 
firewall-cmd --add-port=15672/tcp --permanent
# 重新载入
firewall-cmd --reload
# 再次查询，结果就是 yes 了
firewall-cmd --query-port=15672/tcp

# 新增用户 用户名和密码都是 admin
rabbitmqctl add_user admin admin

# 设置用户分配操作权限，admin 用户的权限为 administrator
rabbitmqctl set_user_tags admin administrator

# 命令格式为: set_permissions [-p <vhostpath>] <user> <conf> <write> <read>
# 这里即为 admin 用户开启 配置文件和读写的权限
rabbitmqctl set_permissions -p / admin ".*"".*"".*"

# 更新 yum
yum update
# 检查yum依赖的几个包 yum-utils 提供 yum-config-manager 功能， 后面两个是 devicemapper 用到的
yum install -y yum-utils device-mapper-persistent-data lvm2

yum-config-manager --add-repo http://mirrors.aliyun.com/docker-ce/linux/centos/docker-ce.repo

yum install docker-ce -y

docker -v

sudo mkdir -p /etc/docker
sudo tee /etc/docker/daemon.json <<-'eof'
{
  "registry-mirrors": ["https://<你的id>.mirror.aliyuncs.com"]
}
eof
sudo systemctl daemon-reload
sudo systemctl restart docker

# 启动 docker
systemctl docker start
# 停止 docker
systemctl docker stop
# 重启 docker
systemctl docker restart
# 查看 docker 状态
systemctl status docker
# 开机自启
systemctl enable docker
systemctl unenable docker

# 导入镜像文件
docker load < xxx.tar.gz
# 查看安装的镜像
docker images
# 删除镜像
docker rmi 镜像名

docker pull rabbitmq:management

docker run -id --name 容器名 -p 15672:15672 -p 5672:5672 rabbitmq:management

docker run -di --name myrabbitmq -e rabbitmq_default_user=admin -e rabbitmq_default_pass=admin -p 15672:15672 -p
 5672:5672 -p 25672:25672 -p 61613:61613 -p 1883:1883 rabbitmq:management

# 查看容器运行状态
docker ps -a
# 启动
docker start 容器名
# 停止
docker stop 容器名
# 退出命令行，不停止
exit
# 进入到node容器（如果开启了 -t 的情况）
docker exec -it 容器名 bash

# 查询 15672 是否开放，一般默认都是 no
firewall-cmd --query-port=15672/tcp
# 开放指定端口 15672 
firewall-cmd --add-port=15672/tcp --permanent
# 重新载入
firewall-cmd --reload
# 再次查询，结果就是 yes 了
firewall-cmd --query-port=15672/tcp

systemctl disable firewalld
systemctl stop firewalld   

<dependency>
   <groupid>com.rabbitmq</groupid>
    <artifactid>amqp-client</artifactid>
    <version>5.10.0</version>
</dependency>
<dependency>
    <groupid>junit</groupid>
    <artifactid>junit</artifactid>
    <version>4.11</version>
</dependency>

host=192.168.122.1
port=5672
virtualhost=/rabbitmq_maven_01
username=admin
password=adminv

public class producer {
    @test
    public void sendmessage() throws ioexception, timeoutexception {
        // 通过工具类获取连接
        connection connection = rabbitmqutil.getconnection();
        // 获取连接通道
        channel channel = connection.createchannel();
        // 通道绑定消息队列
        channel.queuedeclare("queue1",false,false,false,null);
        // 发布消息
        channel.basicpublish("","queue1",null,"this is rabbitmq message 001 !".getbytes());
        // 通过工具关闭channel和释放连接
        rabbitmqutil.close(channel,connection);
    }
}public class producer {
    @test
    public void sendmessage() throws ioexception, timeoutexception {
        // 通过工具类获取连接
        connection connection = rabbitmqutil.getconnection();
        // 获取连接通道
        channel channel = connection.createchannel();
        // 通道绑定消息队列
        channel.queuedeclare("queue1",false,false,false,null);
        // 发布消息
        channel.basicpublish("","queue1",null,"this is rabbitmq message 001 !".getbytes());
        // 通过工具关闭channel和释放连接
        rabbitmqutil.close(channel,connection);
    }
}

public class consumer {
    public static void main(string[] args) throws ioexception, timeoutexception{
        // 通过工具类获取连接
        connection connection = rabbitmqutil.getconnection();
        // 获取连接通道
        channel channel = connection.createchannel();
        // 通道绑定消息队列
        channel.queuedeclare("queue1", false, false, false, null);
        // 消费消息
        channel.basicconsume("queue1", true, new defaultconsumer(channel) {
            @override
            public void handledelivery(string consumertag, envelope envelope, amqp.basicproperties properties, byte[] body) throws ioexception {
                system.out.println("new string(body): " + new string(body));
            }
        });
    }
}

public class producer {
    @test
    public void sendmessage() throws ioexception, timeoutexception {
        // 通过工具类获取连接
        connection connection = rabbitmqutil.getconnection();
        // 获取连接通道
        channel channel = connection.createchannel();
        // 通道绑定消息队列
        channel.queuedeclare("work", true, false, false, null);
        for (int i = 1; i <= 20; i++) {
            // 发布消息
            channel.basicpublish("", "work", null, (i + "号消息").getbytes());
        }
        // 通过工具关闭channel和释放连接
        rabbitmqutil.close(channel, connection);
    }
}

public class consumer1 {
    public static void main(string[] args) throws ioexception {
        // 通过工具类获取连接
        connection connection = rabbitmqutil.getconnection();
        // 获取连接通道
        final channel channel = connection.createchannel();
        // 通道绑定消息队列
        channel.queuedeclare("work", true, false, false, null);
        // 消费消息
        channel.basicconsume("work", true, new defaultconsumer(channel) {
            @override
            public void handledelivery(string consumertag, envelope envelope, amqp.basicproperties properties, byte[] body) throws ioexception {
                try {
                    thread.sleep(2000);
                } catch (interruptedexception e) {
                    e.printstacktrace();
                }
                system.out.println("消费者1号：消费-" + new string(body));
            }
        });
    }
}

public class consumer2 {
    public static void main(string[] args) throws ioexception {
        // 通过工具类获取连接
        connection connection = rabbitmqutil.getconnection();
        // 获取连接通道
        final channel channel = connection.createchannel();
        // 通道绑定消息队列
        channel.queuedeclare("work", true, false, false, null);
        // 消费消息
        channel.basicconsume("work", true, new defaultconsumer(channel) {
            @override
            public void handledelivery(string consumertag, envelope envelope, amqp.basicproperties properties, byte[] body) throws ioexception {
                system.out.println("消费者2号：消费-" + new string(body));
            }
        });
    }

public class producer {
    @test
    public void sendmessage() throws ioexception, timeoutexception {
        // 通过工具类获取连接
        connection connection = rabbitmqutil.getconnection();
        // 获取连接通道
        channel channel = connection.createchannel();
        // 一次只发送一条消息
        channel.basicqos(1);
        // 通道绑定消息队列
        channel.queuedeclare("work", true, false, false, null);
        for (int i = 1; i <= 20; i++) {
            // 发布消息
            channel.basicpublish("", "work", null, (i + "号消息").getbytes());
        }
        // 通过工具关闭channel和释放连接
        rabbitmqutil.close(channel, connection);
    }

public class consumer1 {
    public static void main(string[] args) throws ioexception {
        // 通过工具类获取连接
        connection connection = rabbitmqutil.getconnection();
        // 获取连接通道
        final channel channel = connection.createchannel();
        // 一次只接受一条未确认的消息
        channel.basicqos(1);
        // 通道绑定消息队列
        channel.queuedeclare("work", true, false, false, null);
        // 消费消息
        channel.basicconsume("work", false, new defaultconsumer(channel) {
            @override
            public void handledelivery(string consumertag, envelope envelope, amqp.basicproperties properties, byte[] body) throws ioexception {
                try {
                    thread.sleep(2000);
                } catch (interruptedexception e) {
                    e.printstacktrace();
                }
                system.out.println("消费者1号：消费-" + new string(body));
                // 返回 deliverytag 代表队列可以删除此消息了
                channel.basicack(envelope.getdeliverytag(), false);
            }
        });
    }
}

public class consumer2 {
    public static void main(string[] args) throws ioexception {
        // 通过工具类获取连接
        connection connection = rabbitmqutil.getconnection();
        // 获取连接通道
        final channel channel = connection.createchannel();
        //步骤一:一次只接受一条未确认的消息
        channel.basicqos(1);
        // 通道绑定消息队列
        channel.queuedeclare("work", true, false, false, null);
        // 消费消息
        channel.basicconsume("work", false, new defaultconsumer(channel) {
            @override
            public void handledelivery(string consumertag, envelope envelope, amqp.basicproperties properties, byte[] body) throws ioexception {
                system.out.println("消费者2号：消费-" + new string(body));
                channel.basicack(envelope.getdeliverytag(), false);
            }
        });
    }
public class consumer2 {
    public static void main(string[] args) throws ioexception {
        // 通过工具类获取连接
        connection connection = rabbitmqutil.getconnection();
        // 获取连接通道
        final channel channel = connection.createchannel();
        //步骤一:一次只接受一条未确认的消息
        channel.basicqos(1);
        // 通道绑定消息队列
        channel.queuedeclare("work", true, false, false, null);
        // 消费消息
        channel.basicconsume("work", false, new defaultconsumer(channel) {
            @override
            public void handledelivery(string consumertag, envelope envelope, amqp.basicproperties properties, byte[] body) throws ioexception {
                system.out.println("消费者2号：消费-" + new string(body));
                channel.basicack(envelope.getdeliverytag(), false);
            }
        });
    }

public class producer {
    @test
    public void sendmessage() throws ioexception, timeoutexception {
        // 通过工具类获取连接
        connection connection = rabbitmqutil.getconnection();
        // 获取连接通道
        final channel channel = connection.createchannel();
        // 声明交换机
        channel.exchangedeclare("order", "fanout");
        for (int i = 1; i <= 20; i++) {
            // 发布消息
            channel.basicpublish("order", "", null, "fanout!".getbytes());
        }
        // 通过工具关闭channel和释放连接
        rabbitmqutil.close(channel, connection);
    }
}

public class consumer1 {
    public static void main(string[] args) throws ioexception {
        // 通过工具类获取连接
        connection connection = rabbitmqutil.getconnection();
        channel channel = connection.createchannel();
        // 声明交换机
        channel.exchangedeclare("order", "fanout");
        // 创建临时队列
        string queue = channel.queuedeclare().getqueue();
        // 绑定临时队列和交换机
        channel.queuebind(queue, "order", "");
        // 消费消息
        channel.basicconsume(queue, true, new defaultconsumer(channel) {
            @override
            public void handledelivery(string consumertag, envelope envelope, amqp.basicproperties properties, byte[] body) throws ioexception {
                system.out.println("消费者1号：消费-" + new string(body));
            }
        });
    }
}

public class consumer2 {
    public static void main(string[] args) throws ioexception {
       // 通过工具类获取连接
        connection connection = rabbitmqutil.getconnection();
        // 获取连接通道
        channel channel = connection.createchannel();
        channel channel2 = connection.createchannel();
        // 声明交换机
        channel.exchangedeclare("order", "fanout");
        channel2.exchangedeclare("order", "fanout");
        // 创建临时队列
        string queue = channel.queuedeclare().getqueue();
        system.out.println(queue);
        // 绑定临时队列和交换机
        channel.queuebind(queue, "order", "");
        channel2.queuebind(queue, "order", "");
        // 消费消息
        channel.basicconsume(queue, true, new defaultconsumer(channel) {
            @override
            public void handledelivery(string consumertag, envelope envelope, amqp.basicproperties properties, byte[] body) throws ioexception {
                system.out.println("消费者2号：消费-" + new string(body));
            }
        });
        // 消费消息
        channel2.basicconsume(queue, true, new defaultconsumer(channel2) {
            @override
            public void handledelivery(string consumertag, envelope envelope, amqp.basicproperties properties, byte[] body) throws ioexception {
                system.out.println("消费者2-2号：消费-" + new string(body));
            }
        });
    }
}

public class producer {
    @test
    public void sendmessage() throws ioexception, timeoutexception {
        // 通过工具类获取连接
        connection connection = rabbitmqutil.getconnection();
        // 获取连接通道
        channel channel = connection.createchannel();
        // 声明交换机
        channel.exchangedeclare("order_direct", "direct");
        // 指定 routingkey 
        string key = "info";
        // 发布消息
        channel.basicpublish("order_direct", key, null, ("发送给指定路由" + key + "的消息").getbytes());
        // 通过工具关闭channel和释放连接
        rabbitmqutil.close(channel, connection);
    }
}

public class consumer1 {
    public static void main(string[] args) throws ioexception {
        // 通过工具类获取连接
        connection connection = rabbitmqutil.getconnection();
        channel channel = connection.createchannel();
        // 声明交换机
        channel.exchangedeclare("order_direct", "direct");
        // 获取临时队列
        string queue = channel.queuedeclare().getqueue();
        // 绑定临时队列和交换机
        channel.queuebind(queue, "order_direct", "info");
        channel.queuebind(queue, "order_direct", "error");
        channel.queuebind(queue, "order_direct", "warn");
        // 消费消息
        channel.basicconsume(queue, true, new defaultconsumer(channel) {
            @override
            public void handledelivery(string consumertag, envelope envelope, amqp.basicproperties properties, byte[] body) throws ioexception {
                system.out.println("消费者1：消费-" + new string(body));
            }
        });
    }
}

public class consumer2 {
    public static void main(string[] args) throws ioexception {
         // 通过工具类获取连接
        connection connection = rabbitmqutil.getconnection();
        channel channel = connection.createchannel();
        // 声明交换机
        channel.exchangedeclare("order_direct", "direct");
        // 获取临时队列
        string queue = channel.queuedeclare().getqueue();
        // 绑定临时队列和交换机
        channel.queuebind(queue, "order_direct", "error");
        // 消费消息
        channel.basicconsume(queue, true, new defaultconsumer(channel) {
            @override
            public void handledelivery(string consumertag, envelope envelope, amqp.basicproperties properties, byte[] body) throws ioexception {
                system.out.println("消费者2：消费-" + new string(body));
            }
        });
    }
}

public class producer {
    @test
    public void sendmessage() throws ioexception, timeoutexception {
        // 通过工具类获取连接
        connection connection = rabbitmqutil.getconnection();
        // 获取连接通道
        channel channel = connection.createchannel();
        channel.exchangedeclare("order_topic", "topic");
        // 声明交换机
        string key = "user.query.all";
        // 发布消息
        channel.basicpublish("order_topic", key, null, ("发送给指定路由" + key + "的消息").getbytes());
        rabbitmqutil.close(channel, connection);
    }
}

public class producer {
    @test
    public void sendmessage() throws ioexception, timeoutexception {
        // 通过工具类获取连接
        connection connection = rabbitmqutil.getconnection();
        // 获取连接通道
        channel channel = connection.createchannel();
        channel.exchangedeclare("order_topic", "topic");
        // 声明交换机
        string key = "user.query.all";
        // 发布消息
        channel.basicpublish("order_topic", key, null, ("发送给指定路由" + key + "的消息").getbytes());
        rabbitmqutil.close(channel, connection);
    }
}

public class producer {
    @test
    public void sendmessage() throws ioexception, timeoutexception {
        // 通过工具类获取连接
        connection connection = rabbitmqutil.getconnection();
        // 获取连接通道
        channel channel = connection.createchannel();
        channel.exchangedeclare("order_topic", "topic");
        // 声明交换机
        string key = "user.query.all";
        // 发布消息
        channel.basicpublish("order_topic", key, null, ("发送给指定路由" + key + "的消息").getbytes());
        rabbitmqutil.close(channel, connection);
    }
}

<dependency>
  <groupid>org.springframework.boot</groupid>
  <artifactid>spring-boot-starter-amqp</artifactid>
</dependency>

spring:
  rabbitmq:
    host: 192.168.122.1 # 服务器地址
    port: 5672 # tcp端口
    username: admin # 用户名
    password: admin # 用户密码
    virtual-host: /rabbitmq_springboot_01 # 虚拟主机

@springboottest(classes = rabbitmqspringbootapplication.class)
@runwith(springrunner.class)
public class rabbitmqtest {
    /**
     * 注入 rabbittemplate
     */
    @autowired
    private rabbittemplate rabbittemplate;
    @test
    public void testsimplesendmessage() {
        rabbittemplate.convertandsend("simple_queue", "this is a message !");
    }
}

// 注入容器
@component
// 监听 rabbitmq
@rabbitlistener(queuestodeclare = @queue(value = "simple_queue", durable = "true", exclusive = "false", autodelete = "false"))
public class simpleconsumer {
    // 自动回调
    @rabbithandler
    public void receivemessage(string message) {
        system.out.println("消费者：" + message);
    }
}

spring:
  rabbitmq:
    host: 192.168.122.1 # 服务器地址
    port: 5672 # tcp端口
    username: admin # 用户名
    password: admin # 用户密码
    virtual-host: /rabbitmq_springboot_01 # 虚拟主机
	# 新增部分
    listener:
      simple:
        acknowledge-mode: manual # 开启 ack 手动应答
        prefetch: 1 # 每次只能消费 1 条消息

@component
public class workconsumer {
	// 监听 rabbitmq
    @rabbitlistener(queuestodeclare = @queue("work_queue"))
    // 消费者 1
    public void receivemessage(string body, message message, channel channel) throws ioexception {
        try {
            // 打印输出消息主题
            system.out.println("消费者1：" + body);
            // 返回 deliverytag 代表队列可以删除此消息了
            channel.basicack(message.getmessageproperties().getdeliverytag(),false);
        } catch (ioexception e) {
            e.printstacktrace();
            // 消费者告诉队列信息消费失败
            channel.basicnack(message.getmessageproperties().getdeliverytag(), false, true);
        }
    }
   // 监听 rabbitmq
    @rabbitlistener(queuestodeclare = @queue("work_queue"))
    // 消费者 2
    public void receivemessage2(string body, message message, channel channel) throws ioexception{
        try {
            // 延迟 2s 代表处理业务慢
            thread.sleep(2000);
        } catch (interruptedexception e) {
            e.printstacktrace();
        }
        try {
			// 打印输出消息主题
            system.out.println("消费者2：" + body);
            // 返回 deliverytag 代表队列可以删除此消息了
            channel.basicack(message.getmessageproperties().getdeliverytag(), false);
        } catch (ioexception e) {
            e.printstacktrace();
            // 消费者告诉队列信息消费失败
            channel.basicnack(message.getmessageproperties().getdeliverytag(), false, true);
        }
    }
}

# 发送确认
spring.rabbitmq.publisher-confirm-type=correlated
# spring.rabbitmq.publisher-confirms=true(旧版)
# 发送回调
spring.rabbitmq.publisher-returns=true
# 消费手动确认
spring.rabbitmq.listener.direct.acknowledge-mode=manual
spring.rabbitmq.listener.simple.acknowledge-mode=manual
# 并发消费者初始化值
spring.rabbitmq.listener.simple.concurrency=1
# 并发消费者的最大值
spring.rabbitmq.listener.simple.max-concurrency=10
# 每个消费者每次监听时可拉取处理的消息数量
# 在单个请求中处理的消息个数，他应该大于等于事务数量(unack的最大数量)
spring.rabbitmq.listener.simple.prefetch=1
# 是否支持重试
spring.rabbitmq.listener.simple.retry.enabled=true

@rabbitlistener(queuestodeclare = @queue("work_queue"))
// 将上面的监听，增加 containerfactory 属性，然后将配置好的工厂传入
@rabbitlistener(queuestodeclare = @queue("work_queue"), containerfactory = "worklistenerfactory")

@springboottest(classes = rabbitmqspringbootapplication.class)
@runwith(springrunner.class)
public class rabbitmqtest {
    /**
     * 注入 rabbittemplate
     */
    @autowired
    @test
    public void testfanoutsendmessage() {
        rabbittemplate.convertandsend("order_exchange", "", "this is a message !");
    }
}

@component
public class fanoutconsumer {
    // 绑定临时队列和交换机
    @rabbitlistener(bindings = {
            @queuebinding(
                value = @queue(), // 临时队列
                exchange = @exchange(name = "order_exchange", type = "fanout") // 交换机与类型
            )
    })
    public void receivemessage1(string message) {
        system.out.println("消费者1：" + message);
    }
    // 绑定临时队列和交换机
    @rabbitlistener(bindings = {
            @queuebinding(
                value = @queue(), // 临时队列
                exchange = @exchange(name = "order_exchange", type = "fanout") // 交换机与类型
            )
    })
    public void receivemessage2(string message) {
        system.out.println("消费者2：" + message);
    }
}

@component
public class directconsumer {
    // 绑定临时队列和交换机
    @rabbitlistener(bindings = {
            @queuebinding(
                    value = @queue(), // 临时队列
                    exchange = @exchange(name = "direct_exchange", type = "direct"), // 交换机和类型
                    key = {"info", "warn", "error"} // 路由key
            )
    })
    public void receivemessage1(string message) {
        system.out.println("消费者1：" + message);
    }
     // 绑定临时队列和交换机
    @rabbitlistener(bindings = {
            @queuebinding(
                    value = @queue(), // 临时队列
                    exchange = @exchange(name = "direct_exchange", type = "direct"), // 交换机和类型
                    key = {"info", "warn", "error"} // 路由key
            )
    })
    public void receivemessage2(string message) {
        system.out.println("消费者2：" + message);
    }
}

@component
public class topicconsumer {
    // 绑定临时队列和交换机
    @rabbitlistener(bindings = {
            @queuebinding(
                    value = @queue(), // 临时队列
                    exchange = @exchange(name = "topic_exchange", type = "topic"), // 交换机和类型
                    key = {"order.*"} // 通配符路由key
            )
    })
    public void receivemessage1(string message) {
        system.out.println("消费者1：" + message);
    }
    // 绑定临时队列和交换机
    @rabbitlistener(bindings = {
            @queuebinding(
                    value = @queue(), // 临时队列
                    exchange = @exchange(name = "topic_exchange", type = "topic"), // 交换机和类型
                    key = {"order.*"} // 通配符路由key
            )
    })
    public void receivemessage2(string message) {
        system.out.println("消费者2：" + message);
    }
}

@configuration
public class rabbitmqconfiguration {
    public static final string topic_exchange = "topic_order_exchange";
    public static final string topic_queue_name_1 = "test_topic_queue_1";
    public static final string topic_queue_name_2 = "test_topic_queue_2";
    public static final string topic_routingkey_1 = "test.*";
    public static final string topic_routingkey_2 = "test.#";
    @bean
    public topicexchange topicexchange() {
        return new topicexchange(topic_exchange);
    }
    @bean
    public queue topicqueue1() {
        return new queue(topic_queue_name_1);
    }
    @bean
    public queue topicqueue2() {
        return new queue(topic_queue_name_2);
    }
    @bean
    public binding bindingtopic1(){
        return bindingbuilder.bind(topicqueue1())
                .to(topicexchange())
                .with(topic_routingkey_1);
    }
    @bean
    public binding bindingtopic2(){
        return bindingbuilder.bind(topicqueue2())
                .to(topicexchange())
                .with(topic_routingkey_2);
    }
}

@component
public class topicconsumer {
	// 绑定队列即可
    @rabbitlistener(queues = {rabbitmqconfiguration.topic_queue_name_1})
    public void receivemessage1(string message) {
        system.out.println("消费者1：" + message);
    }
    // 绑定队列即可
    @rabbitlistener(queues = {rabbitmqconfiguration.topic_queue_name_2})
    public void receivemessage2(string message) {
        system.out.println("消费者2：" + message);
    }
}