RabbitMq的安装以及使用-Toy模板网

这篇具有很好参考价值的文章主要介绍了RabbitMq的安装以及使用。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

一，mq的介绍

MQ全称 Message Queue（[kjuː]）（消息队列），是在消息的传输过程中保存消息的容器。多用于分布式系统之间进行通信

二，MQ的优势和劣势

1.劣势

系统可用性降低
系统引入的外部依赖越多，系统稳定性越差。一旦 MQ 宕机，就会对业务造成影响。如何保证MQ的高可用？

系统复杂度提高
MQ 的加入大大增加了系统的复杂度，以前系统间是同步的远程调用，现在是通过 MQ 进行异步调用。如何保证消息没有被重复消费？怎么处理消息丢失情况？那么保证消息传递的顺序性？

一致性问题
A 系统处理完业务，通过 MQ 给B、C、D三个系统发消息数据，如果 B 系统、C 系统处理成功，D 系统处理失败。如何保证消息数据处理的一致性？

2.优势

应用解耦：提高系统容错性和可维护性。
异步提速：提升用户体验和系统吞吐量。
削峰填谷：提高系统稳定性

3.既然MQ有优势也有劣势，那么使用MQ时需要满足那些条件呢？

消费者--》生产者

生产者不需要从消费者处获得反馈。引入消息队列之前的直接调用，其接口的返回值应该为空，这才让明明下层的动作还没做，上层却当成动作做完了继续往后走，即所谓异步成为了可能。

订单->库存

容许短暂的不一致性。
确实是用了有效果。即解耦、提速、削峰这些方面的收益，超过加入MQ，管理MQ这些成本。

三.常见的MQ产品

目前业界有很多的 MQ 产品，例如 RabbitMQ、RocketMQ、ActiveMQ、Kafka、ZeroMQ、MetaMq等，也有直接使用 Redis 充当消息队列的案例，而这些消息队列产品，各有侧重，在实际选型时，需要结合自身需求及 MQ 产品特征，综合考虑。

RabbitMq的安装以及使用,java-rabbitmq,rabbitmq,java

四.RabbitMQ基本介绍

AMQP，即 Advanced Message Queuing Protocol（英[ˈprəʊtəkɒl]）（高级消息队列协议），是一个网络协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。2006年，AMQP 规范发布。类比HTTP。

RabbitMq的安装以及使用,java-rabbitmq,rabbitmq,java

消息队列中间件

exchange 交换机分发消息分发到不同的容器 queue 通过路由来处理

queue 容器

routes 路由

生产者发布消息到exchange exchange 通过不同的路由规则发布/路由给不同的queue 进行存储 cunsumer通过队列去监听拿到消息进行消费

2007年，Rabbit 技术公司基于 AMQP 标准开发的 RabbitMQ 1.0 发布。RabbitMQ 采用 Erlang 语言开发。Erlang 语言由 Ericson 设计，专门为开发高并发和分布式系统的一种语言，在电信领域使用广泛。

1.RabbitMQ基础架构

Broker 中间者服务

procedure 和consumer都是客户端

客户端通过链接和服务端进行通信所以需要建立起来连接然后进行通信a

使用channel（管道）节省资源

一个rabbitmq里面有很多的虚拟机相当于mysql里面有很多数据库，数据库里面有很多表，都是独立的。

每个虚拟机里面有很多的exchange和queue 独立分区的作用

RabbitMq的安装以及使用,java-rabbitmq,rabbitmq,java

2. RabbitMQ 中的相关概念

Broker：接收和分发消息的应用，RabbitMQ Server就是 Message Broker。

Virtual host：出于多租户和安全因素设计的，把 AMQP 的基本组件划分到一个虚拟的分组中，类似于网络中的 namespace 概念。当多个不同的用户使用同一个 RabbitMQ server 提供的服务时，可以划分出多个vhost，每个用户在自己的 vhost 创建 exchange／queue 等。

Connection：publisher／consumer 和 broker 之间的 TCP 连接。

Channel：如果每一次访问 RabbitMQ 都建立一个 Connection，在消息量大的时候建立 TCP Connection的开销将是巨大的，效率也较低。Channel 是在 connection 内部建立的逻辑连接，如果应用程序支持多线程，通常每个thread创建单独的 channel 进行通讯，AMQP method 包含了channel id 帮助客户端和message broker 识别 channel，所以 channel 之间是完全隔离的。Channel 作为轻量级的 Connection 极大减少了操作系统建立 TCP connection 的开销。

Exchange：message 到达 broker 的第一站，根据分发规则，匹配查询表中的 routing key，分发消息到queue 中去。常用的类型有：

direct (point-to-point)

topic (publish-subscribe)

fanout (multicast)

Queue：消息最终被送到这里等待 consumer 取走

Binding：exchange 和 queue 之间的虚拟连接，binding 中可以包含 routing key。Binding 信息被保存到 exchange 中的查询表中，用于 message 的分发依据

3. RabbitMQ的6 种工作模式

RabbitMQ 提供了 6 种工作模式：

简单模式、work queues、Publish/Subscribe 发布与订阅模式、Routing 路由模式、Topics 主题模式、RPC 远程调用模式（远程调用，不太算 MQ；暂不作介绍）。==

官网对应模式介绍：RabbitMQ Tutorials — RabbitMQ

4. AMQP 和 JMS

MQ是消息通信的模型；实现MQ的大致有两种主流方式：AMQP、JMS。

4.1 AMQP

AMQP是一种协议，更准确的说是一种binary wire-level protocol（链接协议）。这是其和JMS的本质差别，AMQP不从API层进行限定，而是直接定义网络交换的数据格式。

4.2 JMS

JMS即Java消息服务（Java Message Service）应用程序接口，是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。

（规定了消息客户端的一套api的东西，rabbitmq没有遵循规则）

JMS 是 JavaEE 规范中的一种，类比JDBC。

4.3 AMQP与 JMS 区别

JMS是定义了统一的接口，来对消息操作进行统一；AMQP是通过规定协议来统一数据交互的格式。

JMS限定了必须使用Java语言；AMQP只是协议，不规定实现方式，因此是跨语言的。

JMS规定了两种消息模式；而AMQP的消息模式更加丰富

4.4 再谈市场上常见的消息队列

ActiveMQ：基于JMS

ZeroMQ：基于C语言开发

RabbitMQ：基于AMQP协议，erlang语言开发，稳定性好

RocketMQ：基于JMS，阿里巴巴产品

Kafka：类似MQ的产品；分布式消息系统，高吞吐量。

五，RabbitMQ的安装和配置

1.安装包准备

2.安装依赖环境

yum -y install build-essential openssl openssl-devel unixODBC unixODBC-devel make gcc gcc-c++ kernel-devel m4 ncurses-devel tk tc xz

3.安装Erlang

rpm -ivh erlang-22.0.7-1.el7.x86_64.rpm

4.安装RabbitMQ

1.安装依赖的包

rpm -ivh socat-1.7.3.2-2.el7.x86_64.rpm

2.安装rabbitmq

rpm -ivh rabbitmq-server-3.7.18-1.el7.noarch.rpm

5. 开启管理界面及配置

rabbitmq-plugins enable rabbitmq_management

RabbitMq的安装以及使用,java-rabbitmq,rabbitmq,java

6.修改默认配置信息

vim /usr/lib/rabbitmq/lib/rabbitmq_server-3.7.18/ebin/rabbit.app

RabbitMq的安装以及使用,java-rabbitmq,rabbitmq,java

7.启动RabbitMQ

systemctl start rabbitmq-server # 启动服务
systemctl stop rabbitmq-server # 停止服务
systemctl restart rabbitmq-server # 重启服务
systemctl status rabbitmq-server #查看状态

8.打开客户端

浏览器上直接查询

192.168.146.254：15672

使用guest/guest登录之后出现如下即为安装成功

RabbitMq的安装以及使用,java-rabbitmq,rabbitmq,java

六、添加用户

RabbitMq的安装以及使用,java-rabbitmq,rabbitmq,java

1.给用户添加权限

RabbitMq的安装以及使用,java-rabbitmq,rabbitmq,java

七RabbitMQ工程搭建

1.简单模式

1.添加pom

<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.6.0</version>
</dependency>

2.生产者

public class productTest {
    @Test
    public void product() throws Exception {
        ConnectionFactory continuation = new ConnectionFactory();
        continuation.setHost("192.168.64.33");
        continuation.setPort(5672);
        continuation.setPassword("ww");
        continuation.setUsername("ww");
//        设置对应的虚拟主机
        continuation.setVirtualHost("/ww");
//        物理连接
        Connection connection = continuation.newConnection();
//        建立通道
        Channel channel = connection.createChannel();
        /**
         * String queue, 队列的名称
         * boolean durable, 持久化
         * boolean exclusive, 是否独占
         * boolean autoDelete,  受否自动删除
         * Map<String, Object> arguments  参数
         */
//        创建队列
        channel.queueDeclare("test1",false,false,false,null);
//        发布消息
        channel.basicPublish("","test1",null,"hello".getBytes());
    }
}

以下就是生产消息完成

RabbitMq的安装以及使用,java-rabbitmq,rabbitmq,java

3.消费者

public class Consumer {
    @Test
    public void product() throws Exception {
        ConnectionFactory continuation = new ConnectionFactory();
        continuation.setHost("192.168.64.33");
        continuation.setPort(5672);
        continuation.setPassword("ww");
        continuation.setUsername("ww");
//        设置对应的虚拟主机
        continuation.setVirtualHost("/ww");
//        物理连接
        Connection connection = continuation.newConnection();
//        建立通道
        Channel channel = connection.createChannel();
 
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String s = new String(body);
                System.out.println("消息队列:"+s);
            }
        };
        channel.basicConsume("test1",true,defaultConsumer);
    }
}

以下就是消费者获取生产者发布的消息

RabbitMq的安装以及使用,java-rabbitmq,rabbitmq,java

2.竞争消费者模式

1.生产者

public class productTest1 {
    @Test
    public void product() throws Exception {
        ConnectionFactory continuation = new ConnectionFactory();
        continuation.setHost("192.168.64.33");
        continuation.setPort(5672);
        continuation.setPassword("ww");
        continuation.setUsername("ww");
//        设置对应的虚拟主机
        continuation.setVirtualHost("/ww");
//        物理连接
        Connection connection = continuation.newConnection();
//        建立通道
        Channel channel = connection.createChannel();
        /**
         * String queue, 队列的名称
         * boolean durable, 持久化
         * boolean exclusive, 是否独占
         * boolean autoDelete,  受否自动删除
         * Map<String, Object> arguments  参数
         */
//        创建队列
        channel.queueDeclare("test1",false,false,false,null);
 
        for (int i = 0; i < 10; i++) {
            //        发布消息
            channel.basicPublish("","test1",null,("hello"+i).getBytes());
        }
    }
}

2.消费者

1.消费者1

public class Consumer1 {
    public static void main(String[] args) throws Exception {
        ConnectionFactory continuation = new ConnectionFactory();
        continuation.setHost("192.168.64.33");
        continuation.setPort(5672);
        continuation.setPassword("ww");
        continuation.setUsername("ww");
//        设置对应的虚拟主机
        continuation.setVirtualHost("/ww");
//        物理连接
        Connection connection = continuation.newConnection();
//        建立通道
        Channel channel = connection.createChannel();
 
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String s = new String(body);
                System.out.println("消息队列:" + s);
            }
        };
        channel.basicConsume("test1", true, defaultConsumer);
    }
}

2.消费者

public class Consumer2 {
    public static void main(String[] args) throws Exception {
        ConnectionFactory continuation = new ConnectionFactory();
        continuation.setHost("192.168.64.33");
        continuation.setPort(5672);
        continuation.setPassword("ww");
        continuation.setUsername("ww");
//        设置对应的虚拟主机
        continuation.setVirtualHost("/ww");
//        物理连接
        Connection connection = continuation.newConnection();
//        建立通道
        Channel channel = connection.createChannel();
 
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String s = new String(body);
                System.out.println("消息队列:"+s);
            }
        };
        channel.basicConsume("test1",true,defaultConsumer);
    }
    }

项目结构

RabbitMq的安装以及使用,java-rabbitmq,rabbitmq,java

3.订阅模式

1.生产者

public class productTest2 {
    @Test
    public void product() throws Exception {
        ConnectionFactory continuation = new ConnectionFactory();
        continuation.setHost("192.168.64.33");
        continuation.setPort(5672);
        continuation.setPassword("ww");
        continuation.setUsername("ww");
//        设置对应的虚拟主机
        continuation.setVirtualHost("/ww");
//        物理连接
        Connection connection = continuation.newConnection();
//        建立通道
        Channel channel = connection.createChannel();
//        创建交换机
        channel.exchangeDeclare("myjhj", BuiltinExchangeType.FANOUT,false);
//        创建队列
        channel.queueDeclare("testj3",false,false,false,null);
        channel.queueDeclare("testj4",false,false,false,null);
//        交换机绑定
        channel.queueBind("testj3","myjhj","");
        channel.queueBind("testj4","myjhj","");
//        消息
        channel.basicPublish("myjhj","",null,"testex".getBytes());
    }
}

消费者的同上

4.路由模式

1.生产者

public class productTest3 {
    @Test
    public void product() throws Exception {
        ConnectionFactory continuation = new ConnectionFactory();
        continuation.setHost("192.168.64.33");
        continuation.setPort(5672);
        continuation.setPassword("ww");
        continuation.setUsername("ww");
//        设置对应的虚拟主机
        continuation.setVirtualHost("/ww");
//        物理连接
        Connection connection = continuation.newConnection();
//        建立通道
        Channel channel = connection.createChannel();
//        创建交换机
        channel.exchangeDeclare("myjhj1", BuiltinExchangeType.DIRECT,false);
//        创建队列
        channel.queueDeclare("testj5",false,false,false,null);
        channel.queueDeclare("testj6",false,false,false,null);
//        交换机绑定队列
        channel.queueBind("testj5","myjhj1","error");
        channel.queueBind("testj6","myjhj1","test6666");
        channel.queueBind("testj6","myjhj1","test7777");
//        消息
        channel.basicPublish("myjhj1","test7777",null,"测试".getBytes());
    }
}

消费者同上

5.通配符模式

Topic类型与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！

通配符规则：

#：匹配一个或多个词

*：匹配不多不少恰好1个词 test.* test.insert

Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert

举例：

item.#：能够匹配item.insert.abc 或者 item.insert

item.*：只能匹配item.insert

1.生产者

public class productTest3 {
    @Test
    public void product() throws Exception {
        ConnectionFactory continuation = new ConnectionFactory();
        continuation.setHost("192.168.64.33");
        continuation.setPort(5672);
        continuation.setPassword("ww");
        continuation.setUsername("ww");
//        设置对应的虚拟主机
        continuation.setVirtualHost("/ww");
//        物理连接
        Connection connection = continuation.newConnection();
//        建立通道
        Channel channel = connection.createChannel();
//        创建交换机
        channel.exchangeDeclare("myjhj3", BuiltinExchangeType.TOPIC,false);
//        创建队列
        channel.queueDeclare("testj9",false,false,false,null);
        channel.queueDeclare("testj10",false,false,false,null);
//        交换机绑定队列
        channel.queueBind("testj9","myjhj3","test.#");
        channel.queueBind("testj10","myjhj3","test.*");
        channel.queueBind("testj10","myjhj3","*.aaa");
//        消息
        //受作用的为9和10
        channel.basicPublish("myjhj3","test.aaa",null,"测试111".getBytes());
    }
}

消费者同上文章来源地址https://www.toymoban.com/news/detail-774332.html

到了这里，关于RabbitMq的安装以及使用的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！