深入学习RabbitMQ五种模式(一)

简介: 深入学习RabbitMQ五种模式(一)

1.安装erlang

下载otp_win64_25.3.exe

https://www.erlang.org/downloads

erlang安装完成,需要配置erlang环境变量

ERLANG_HOME=E:\software\Erlang OTP
PATH=%PATH%;%ERLANG_HOME%\bin;

2.安装RabbitMQ

下载rabbitmq-server-3.11.13.exe

https://www.rabbitmq.com/download.html

进入安装目录下sbin目录,安装并运行服务

安装服务: rabbitmq-service.bat install
删除服务:rabbitmq-service.bat remove
启动服务:rabbitmq-service.bat start
停止服务: rabbitmq-service.bat stop

安装管理插件

安装RabbitMQ的管理插件,方便在浏览器端管理RabbitMQ

管理员身份打开cmd,进入E:\software\RabbitMQ Server\rabbitmq_server-3.11.13\sbin目录,运行

rabbitmq-plugins.bat enable rabbitmq_management

执行结果

重启一下RabbitMQ

启动成功,登录RabbitMQ

访问地址http://127.0.0.1:15672/;初始账号和密码guest/guest

3. RabbitMQ常用五种模式

3.1. 简单模式

该模式是个一对一模式,只有一个生产者Producer(用于生产消息),一个队列Queue(用于存储消息),一个消费者Consumer (用于接收消息)。

注:简单模式也用到了交换机,使用的是默认的交换机(AMQP default)。

  • 创建项目rabbitmq-learn

pom.xml引入以下依赖

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>com.olive</groupId>
  <artifactId>rabbitmq-learn</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <properties>
<maven.compiler.target>1.8</maven.compiler.target>
<maven.compiler.source>1.8</maven.compiler.source>
</properties>
  <dependencies>
    <!-- mq的依赖 -->
    <dependency>
      <groupId>com.rabbitmq</groupId>
      <artifactId>amqp-client</artifactId>
      <version>5.16.0</version>
    </dependency>
    <!-- 日志处理 -->
    <dependency>
      <groupId>org.slf4j</groupId>
      <artifactId>slf4j-log4j12</artifactId>
      <version>1.7.21</version>
    </dependency>
    <dependency>
      <groupId>log4j</groupId>
      <artifactId>log4j</artifactId>
      <version>1.2.17</version>
    </dependency>
  </dependencies>
</project>

RabbitMQ官方提供了amqp-clientjava客户端连接RabbitMQ Server;仓库地址如下

https://github.com/rabbitmq/rabbitmq-java-client
  • RabbitMQ连接的工具类
package com.olive;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/**
 * 封装连接工具类
 */
public class ConnectionUtils {
public static Connection getConnection() throws Exception {
// 1.定义连接工厂
ConnectionFactory factory = new ConnectionFactory();
// 2.设置服务器地址
        factory.setHost("127.0.0.1");
// 3.设置协议端口号
        factory.setPort(5672);
// 4.虚拟主机名称;默认为 /
        factory.setVirtualHost("/");
// 5.设置用户名称
        factory.setUsername("admin");
// 6.设置用户密码
        factory.setPassword("admin123");
// 7.创建连接
Connection connection = factory.newConnection();
return connection;
    }
}

在RabbitMQ管理后台创建admin用户;可以使用默认的guest用户。

  • 创建生产者

生产者负责创建消息并且将消息发送至指定的队列中,简单分为5步:

创建连接 ——> 创建通道 ——> 创建(声明)队列 ——> 发送消息 ——> 关闭资源

package com.olive;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
/**
 * 生产者(简单模式)
 */
public class SimpleProducer {
/**队列名称*/
private static final String QUEUE_NAME = "simple_queue";
public static void main(String[] args) throws Exception {
// 1、获取连接
Connection connection = ConnectionUtils.getConnection();
// 2、创建通道(频道)
Channel channel = connection.createChannel();
// 3、声明(创建)队列
/*
         * queue      参数1:声明通道中对应的队列名称
         * durable    参数2:是否定义持久化队列,当mq重启之后队列还在
         * exclusive  参数3:是否独占本次连接,为true则只能有一个消费者监听这个队列
         * autoDelete 参数4:是否自动删除队列,如果为true表示没有消息也没有消费者连接自动删除队列
         * arguments  参数5:队列其它参数(额外配置)
         */
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
// 4.发送消息
/*
         * exchange   参数1:交换机名称,如果没有指定则使用默认Default Exchange
         * routingKey 参数2:队列名称或者routingKey,如果指定了交换机就是routingKey路由key,简单模式可以传递队列名称
         * props      参数3:消息的配置信息
         * body       参数4:要发送的消息内容
         */
String msg = "Hello World RabbitMQ!!!";
        System.out.println("生产者发送的消息:" + msg);
        channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
//关闭资源
        channel.close();
        connection.close();
    }
}
  • 创建消费者

消费者实现和生产者实现过程差不多,但是没有关闭通道和连接,因为消费者要一直等待随时可能发来的消息,大致分为如下3步:

获取连接 ——> 创建通道 ——> 监听队列,接收消息

package com.olive;
import java.io.IOException;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
/**
 * 消费者(简单模式)
 */
public class SimpleConsumer {
/**队列名称*/
private static final String QUEUE_NAME = "simple_queue";
public static void main(String[] args) throws Exception {
// 1、获取连接对象
Connection connection = ConnectionUtils.getConnection();
// 2、创建通道(频道)
Channel channel = connection.createChannel();
// 3. 创建队列Queue,如果没有一个名字叫simple_world的队列,则会创建该队列,如果有则不会创建.
// 这里可有可无,但是发送消息是必须得有该队列,否则消息会丢失
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
// 4、监听队列,接收消息
DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
/*
             *  handleDelivery回调方法,当收到消息后,会自动执行该方法
             *  consumerTag 参数1:消费者标识
             *  envelope    参数2:可以获取一些信息,如交换机,路由key...
             *  properties  参数3:配置信息
             *  body        参数4:读取到的消息
             */
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者获取消息:" + new String(body));
            }
        };
/*
         * queue    参数1:队列名称
         * autoAck  参数2:是否自动确认,true表示自动确认接收完消息以后会自动将消息从队列移除。否则需要手动ack消息
         * callback 参数3:回调对象,在上面定义了
         */
        channel.basicConsume(QUEUE_NAME, true, defaultConsumer);
//注意,消费者这里不建议关闭资源,让程序一直处于读取消息的状态
    }
}
  • 验证测试

运行生产者的代码,表示向队列中发送消息。

查看RabbitMQ控制台中的Queues内容

启动消费者,消费RabbitMQ队列中的消息。

在通过RabbitMQ控制台查看Queues的内容;发现消息已经被消费

简单模式的不足之处:该模式是一对一,一个生产者向一个队列中发送消息,一个消费者从绑定的队列中获取消息,这样耦合性过高,如果有多个消费者想消费队列中信息就无法实现了。

相关实践学习
RocketMQ一站式入门使用
从源码编译、部署broker、部署namesrv,使用java客户端首发消息等一站式入门RocketMQ。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
2月前
|
消息中间件 Java API
RocketMQ事务消息, 图文、源码学习探究~
介绍 RocketMQ是阿里巴巴开源的分布式消息中间件,它是一个高性能、低延迟、可靠的消息队列系统,用于在分布式系统中进行异步通信。 从4.3.0版本开始正式支持分布式事务消息~ RocketMq事务消息支持最终一致性:在普通消息基础上,支持二阶段的提交能力。将二阶段提交和本地事务绑定,实现全局提交结果的一致性。 原理、流程 本质上RocketMq的事务能力是基于二阶段提交来实现的 在消息发送上,将二阶段提交与本地事务绑定 本地事务执行成功,则事务消息成功,可以交由Consumer消费 本地事务执行失败,则事务消息失败,Consumer无法消费 但是,RocketMq只能保证本地事务
|
3月前
|
物联网 Go 网络性能优化
使用Go语言(Golang)可以实现MQTT协议的点对点(P2P)消息发送。MQTT协议本身支持多种消息收发模式
使用Go语言(Golang)可以实现MQTT协议的点对点(P2P)消息发送。MQTT协议本身支持多种消息收发模式【1月更文挑战第21天】【1月更文挑战第104篇】
101 1
|
20天前
|
消息中间件 存储 负载均衡
消息队列学习之RabbitMQ
【4月更文挑战第3天】消息队列学习之RabbitMQ,一种基于erlang语言开发的流行的开源消息中间件。
15 0
|
29天前
|
消息中间件 Java RocketMQ
【深度挖掘 RocketMQ底层源码】「底层源码挖掘系列」抽丝剥茧贯穿RocketMQ的消费者端的运行核心的流程(Pull模式-下)
【深度挖掘 RocketMQ底层源码】「底层源码挖掘系列」抽丝剥茧贯穿RocketMQ的消费者端的运行核心的流程(Pull模式-下)
12 1
|
29天前
|
消息中间件 存储 NoSQL
【深度挖掘 RocketMQ底层源码】「底层源码挖掘系列」透彻剖析贯穿RocketMQ的消费者端的运行核心的流程(Pull模式-上)
【深度挖掘 RocketMQ底层源码】「底层源码挖掘系列」透彻剖析贯穿RocketMQ的消费者端的运行核心的流程(Pull模式-上)
27 1
|
1月前
|
传感器 监控 网络协议
MQTT 发布、订阅模式介绍
【2月更文挑战第17天】
68 6
MQTT 发布、订阅模式介绍
|
2月前
|
消息中间件 RocketMQ Docker
分布式事物【RocketMQ事务消息、Docker安装 RocketMQ、实现订单微服务、订单微服务业务层实现】(八)-全面详解(学习总结---从入门到深化)
分布式事物【RocketMQ事务消息、Docker安装 RocketMQ、实现订单微服务、订单微服务业务层实现】(八)-全面详解(学习总结---从入门到深化)
53 0
|
3月前
|
消息中间件 网络架构
【面试问题】什么是 MQ topic 交换器(模式匹配) ?
【1月更文挑战第27天】【面试问题】什么是 MQ topic 交换器(模式匹配) ?
|
3月前
|
物联网 Go 网络性能优化
MQTT协议本身支持多种消息收发模式
MQTT协议本身支持多种消息收发模式【1月更文挑战第24天】【1月更文挑战第120篇】
35 3
|
3月前
|
消息中间件 Java Spring
一文看懂Spring Boot整合Rabbit MQ实现多种模式的生产和消费
一文看懂Spring Boot整合Rabbit MQ实现多种模式的生产和消费
66 0

热门文章

最新文章