RabbitMQ入门

本文是作者原创,版权归作者所有.若要转载,请注明出处.

本文RabbitMQ版本为rabbitmq-server-3.7.17,erlang为erlang-22.0.7.请各位去官网查看版本匹配和下载,也可以留言,我发安装包

AMQP协议

AMQP 一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。

AMQP是一个二进制协议，拥有一些现代化特点：多信道、协商式，异步，安全，扩平台，中立，高效。

RabbitMQ是AMQP协议的Erlang的实现。

概念	说明
连接Connection	一个网络连接，比如TCP/IP套接字连接。
会话Session	端点之间的命名对话。在一个会话上下文中，保证“恰好传递一次”。
信道Channel	多路复用连接中的一条独立的双向数据流通道。为会话提供物理传输介质。
客户端Client	AMQP连接或者会话的发起者。AMQP是非对称的，客户端生产和消费消息，服务器存储和路由这些消息。
服务节点Broker	消息中间件的服务节点；一般情况下可以将一个RabbitMQ Broker看作一台RabbitMQ 服务器。
端点	AMQP对话的任意一方。一个AMQP连接包括两个端点（一个是客户端，一个是服务器）。
消费者Consumer	一个从消息队列里请求消息的客户端程序。
生产者Producer	一个向交换机发布消息的客户端应用程序。

简单队列

1.创建工程,添加依赖

<dependencies>
        <!--RabbitMQ的依赖-->
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.6.0</version>
        </dependency>
    </dependencies>

2. 编写生产者

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

public class Producer {

    static final String QUEUE_NAME = "simple_queue";
    static final String Host = "192.168.75.163";
    static final Integer Port = 5672;
    static final String VirtualHost = "/hello";
    static final String Username = "test01";
    static final String Password = "test01";


    public static void main(String[] args) throws Exception{
        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //主机地址：默认为localhost
        connectionFactory.setHost(Host);
        //连接端口
        connectionFactory.setPort(Port);
        //虚拟主机名称：默认为 /
        connectionFactory.setVirtualHost(VirtualHost);
        //连接用户名：默认为guest
        connectionFactory.setUsername(Username);
        //连接密码： 默认为guest
        connectionFactory.setPassword(Password);
        //创建连接
        Connection connection = connectionFactory.newConnection();
        //创建频道
        Channel channel = connection.createChannel();
        //声明（创建）队列
        /*
        参数一：队列名称
        参数二：是否定义持久化队列
        参数三：是否独占本次连接
        参数四：是否在不使用的时候自动删除队列
        参数五：队列其他参数
         */
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);
        //要发送的消息
        String message = "hello，rabbitmq！";
        /**
         * 参数1：交换机名称，不能为null 如果没有指定则使用默认Default Exchage
         * 参数2：路由key,简单模式可以传递队列名称
         * 参数3：消息其它属性
         * 参数4：消息内容
         */
        channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
        System.out.println("已发送消息："+message);
        //释放资源
        channel.close();
        connection.close();
    }

}

我们debug看下消息的产生过程

在这里Connection还没创建

继续下一行

此时Connection创建了,Channel还没创建

继续下一行

此时Channel创建了,Queue还没创建,继续跟

此时Queue创建了,msg还没内容,继续跟

此时msg已经有内容了,然后释放资源,结束.

此时connection和channel都关闭了,只有queue还在,且有一条消息

3. 编写消费者

抽取创建connection的工具类

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

public class ConnectionUtil {

    private static final String QUEUE_NAME = "simple_queue";
    private static final String Host = "192.168.75.163";
    private static final Integer Port = 5672;
    private static final String VirtualHost = "/hello";
    private static final String Username = "test01";
    private static final String Password = "test01";

    public static Connection getConnection() throws Exception {
        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //主机地址;默认为 localhost
        connectionFactory.setHost(Host);
        //连接端口;默认为 5672
        connectionFactory.setPort(Port);
        //虚拟主机名称;默认为 /
        connectionFactory.setVirtualHost(VirtualHost);
        //连接用户名；默认为guest
        connectionFactory.setUsername(Username);
        //连接密码；默认为guest
        connectionFactory.setPassword(Password);
        //创建连接
        return connectionFactory.newConnection();
    }
}

编写消息的消费者

import com.itheima.rabbitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;
import java.io.IOException;

public class Consumer {

    public static void main(String[] args) throws Exception {
        //获取连接
        Connection connection = ConnectionUtil.getConnection();
        //创建频道
        Channel channel = connection.createChannel();
        //创建队列：并设置消息处理
        /**
         * 参数1：队列名称
         * 参数2：是否定义持久化队列
         * 参数3：是否独占本次连接
         * 参数4：是否在不使用的时候自动删除队列
         * 参数5：队列其它参数
         */
        channel.queueDeclare(Producer.QUEUE_NAME,true,false,false,null);
        //监听消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            /*
            consumerTag ：消息者标签，在channel.basicConsume时候可以指定
            envelope: 消息包内容，可从中获取消息id，消息routingkey，交换机，消息和重转标记（收到消息失败后是否需要重新发送）
            properties: 消息属性
            body： 消息
             */
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //路由key
                System.out.println("路由key为：" + envelope.getRoutingKey());
                //交换机
                System.out.println("交换机为：" + envelope.getExchange());
                //消息id
                System.out.println("消息id为：" + envelope.getDeliveryTag());
                //收到的消息
                System.out.println("接收到的消息：" + new String(body, "UTF-8"));
                System.out.println("");
                System.out.println("================================================================");
                System.out.println("");
            }
        };
        /**
         * 参数1：队列名称
         * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
         * 参数3：消息接收到后回调
         */
        channel.basicConsume(Producer.QUEUE_NAME, true, consumer);

        //不应该关闭资源，应该一直监听消息
        //channel.close();
        //connection.close();
    }
}

这里就不debug看了,直接看结果

可以看到,消费者是不应该关闭资源的,并且消息已经消费到了

4. 小结

上述的入门案例中中其实使用的是官网如下的简单模式：

在上图的模型中，有以下概念：

P：生产者，也就是要发送消息的程序
C：消费者：消息的接受者，会一直等待消息到来。
queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。

在此案例中RabbitMQ运转流程：

生产者发送消息
1. 生产者创建连接（Connection），开启一个信道（Channel），连接到RabbitMQ Broker；
2. 声明队列并设置属性；如是否排它，是否持久化，是否自动删除；
3. 将路由键（空字符串）与队列绑定起来；
4. 发送消息至RabbitMQ Broker；
5. 关闭信道；
6. 关闭连接；
消费者接收消息
1. 消费者创建连接（Connection），开启一个信道（Channel），连接到RabbitMQ Broker
2. 向Broker 请求消费相应队列中的消息，设置相应的回调函数；
3. 等待Broker回应闭关投递响应队列中的消息，消费者接收消息；
4. 确认（ack，自动确认）接收到的消息；
5. RabbitMQ从队列中删除相应已经被确认的消息；
6. 关闭信道；
7. 关闭连接；

本文链接：https://www.cnblogs.com/lusaisai/p/12995939.html

RabbitMQ入门

本文是作者原创,版权归作者所有.若要转载,请注明出处.

AMQP协议

简单队列

1.创建工程,添加依赖

2. 编写生产者

3. 编写消费者

4. 小结

RabbitMQ入门的更多相关文章

随机推荐

热门专题

目录导航