服务器之家:专注于服务器技术及软件下载分享
分类导航

PHP教程|ASP.NET教程|Java教程|ASP教程|编程技术|正则表达式|C/C++|IOS|C#|Swift|Android|JavaScript|易语言|

服务器之家 - 编程语言 - Java教程 - RabbitMQ消息中间件示例详解

RabbitMQ消息中间件示例详解

2021-06-22 13:27-Finley- Java教程

这篇文章主要给大家介绍了关于RabbitMQ消息中间件的相关资料,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一学习学习吧

前言

rabbitmq 是使用 erlang 语言开发的消息中间件, 其遵循了高级消息队列协议(advanced message queuing protocol, amqp)。

与 kafka 等消息队列相比,rabbitmq 最大的优势在于其较高的可靠性:

  • 提供确认(ack)和重传机制保证消息完成消费, 消费者异常不会导致消息丢失
  • 提供消息持久化机制, broker 崩溃不会导致消息丢失
  • 集群模式下工作, 保证高可用

因为具有较高可靠性和一致性, rabbitmq 可以胜任订单处理、秒杀等一致性要求较高的业务场景。

rabbitmq 概念与机制

rabbitmq 中的概念模型:

  • broker: 消息中间件实例, 可能是单个节点也可能是运行在多节点集群上的逻辑实体
  • 消息(message): 消息由消息头和消息体两部分组成。消息头中包括routing-key、priority等标准消息头以及其它自定义消息头,用于定义rabbitmq对消息行为。消息体是字节流,包含消息内容。
  • 连接(connection): 客户端与 broker 之间的 tcp连接
  • 信道(channel): channel 是建立在 tcp 连接上的逻辑(虚拟)连接。多个 channel 复用同一个 tcp 连接, 以避免建立 tcp 连接的巨大开销。 rabbitmq 官方要求每个线程使用独立的 channel, 禁止多个线程共用 channel。
  • 生产者(publisher): 发送消息的客户端线程
  • 消费者(consumer): 处理消息的客户端线程
  • 交换机(exchange): 交换机负责将消息投递到相应的队列
  • 队列(queue): 接收并保存交换机投递的消息,直至被消费者成功消费。逻辑结构遵循先进先出fifo。
  • 绑定(binding): 将队列(queue)注册到交换机(exchange)的路由表
  • 虚拟主机(vhost): 每个broker下可建立多个vhost, 每个 vhost 可建立独立的 exchange、queue、绑定及权限系统。同一个 broker 下的 vhost 共享 connection、channel 和 用户系统,就是说可以使用同一个用户身份使用同一个 channel 访问不同 vhost。

交换机(exchange)

生产者发送的消息会首先送到交换机(exchange), 交换机根据自身类型和消息的 routing-key 等信息将消息投递到绑定的消息队列中。

rabbitmq中的四种标准交换机:

direct: 如果消息的 routing-key 与队列的 binding-key 完全相同,direct类型的交换机则会将消息投递到该队列中。

  • 多个队列可以使用相同的 binding-key 绑定到同一个 direct 交换机,direct 交换机会把消息投递到所有 binding-key 与消息 routing-key 相同的队列

topic: 允许队列的 binding-key 中包含通配符*和#, topic 交换机会将消息投递到 binding-key 与 routing-key 匹配的队列中。

  • 通配符按照关键字进行匹配,如news.cn.a中的关键字是news、cn和a,即关键字按照.分割
  • #通配符匹配0个或多个关键字, news.#.a可以匹配news.a, news.cn.a和news.asia.cn.a等
  • *通配符匹配一个关键字, news.*.a匹配news.cn.a不匹配news.a、news.asia.cn.a

fanout: fanout 交换机不进行任何匹配, 将消息投递到所有绑定的队列

header: header 交换机根据消息头进行投递,现在已较少使用

我们可以使用 rabbitmq 的插件机制使用第三方交换机或自行开发交换机。如实现延时投递的。

消息头中的delivery-mode可以设置为 persistent(持久化) 或者 transient(易失)。 exchange 和 queue 在处理持久化的消息时都会先将消息写入磁盘中再进行下一步处理, 即使 rabbitmq 崩溃也不会丢失。

消费者客户端通常使用的channel.basicconsume使用推(push)模式投递消息, 即当有新消息时 broker 通过 channel 主动向客户端发送消息。客户端也可以使用channel.basicget从 broker 拉取消息。

ack机制

rabbitmq 提供了确认送达(acknowledge)机制保证消息被正确处理不会丢失。

确认送达的回执有三种:

  • ack: 消息已被成功处理
  • nack: 消息处理异常, 需要重新投递
  • reject: 消息非法, 丢弃消息

rabbitmq 的 queue 可以设置 no_ack=true, 则消息被投递后即删除不等待回执。

channel.basicconsume 可以指定auto_ack模式,若auto_ack=true当客户端收到完整消息后即会自动发出ack回执,否则必须显式的发出回执。

java 代码示例

首先安装并启动rabbitmq实例, mac用户可以使用 homebrew 进行安装:

?
1
brew install rabbitmq

启动服务:

?
1
brew services start rabbitmq

或者使用官方docker镜像:

?
1
docker run -d --hostname my-rabbit --name some-rabbit rabbitmq:3-management

rabbitmq官网提供了ubuntu、rpm以及windows等多种平台安装方式。

rabbitmq默认tcp端口为5672, web控制台默认端口15672。

在maven中添加依赖:

?
1
2
3
4
5
<dependency>
 <groupid>com.rabbitmq</groupid>
 <artifactid>amqp-client</artifactid>
 <version>5.5.1</version>
</dependency>

编写生产者:

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
package rabbit;
 
import java.io.ioexception;
import java.util.concurrent.timeoutexception;
 
import com.rabbitmq.client.amqp;
import com.rabbitmq.client.channel;
import com.rabbitmq.client.connection;
import com.rabbitmq.client.connectionfactory;
 
/**
 * @author finley
 */
public class rabbitproducer {
 
 public static void main(string[] args) throws ioexception, timeoutexception {
  connectionfactory factory = new connectionfactory();
  factory.setusername("guest");
  factory.setpassword("guest");
  factory.sethost("localhost");
  try (connection conn = factory.newconnection();
    channel channel = conn.createchannel()) {
   string exchangename = "test-exchange";
   channel.exchangedeclare(exchangename, "direct", true);
 
   string routingkey = "hello";
 
   byte[] msg = "hello world".getbytes();
   amqp.basicproperties.builder propsbuilder = new amqp.basicproperties.builder();
   propsbuilder.deliverymode(2); // persistent
   propsbuilder.priority(0); // normal
   propsbuilder.contenttype("text/plain");
   channel.basicpublish(exchangename, routingkey, propsbuilder.build(), msg);
  }
 }
}

编写消费者:

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
package rabbit;
 
import java.io.ioexception;
import java.util.concurrent.timeoutexception;
 
import com.rabbitmq.client.*;
 
/**
 * @author finley
 */
public class rabbitconsumer {
 
 public static void main(string[] args) throws ioexception, timeoutexception {
  connectionfactory factory = new connectionfactory();
  factory.setusername("guest");
  factory.setpassword("guest");
  factory.sethost("localhost");
  try (connection conn = factory.newconnection();
    channel channel = conn.createchannel()) {
   string exchangename = "test-exchange";
   channel.exchangedeclare(exchangename, "direct", true);
 
   string queuename = channel.queuedeclare().getqueue();
   string bindingkey = "hello";
   channel.queuebind(queuename, exchangename, bindingkey);
 
   while(true) {
    channel.basicconsume(queuename, false, "", new defaultconsumer(channel) {
     @override
     public void handledelivery(string consumertag, envelope envelope, amqp.basicproperties properties, byte[] body) throws ioexception {
      string routingkey = envelope.getroutingkey();
      string contenttype = properties.getcontenttype();
      string bodystr = new string(body, "utf-8");
      system.out.println("routingkey: " + routingkey + ", contenttype: " + contenttype + ", body: " + bodystr);
      long deliverytag = envelope.getdeliverytag();
      channel.basicack(deliverytag, false);
     }
    });
   }
  }
 }
 
}

rabbitmq 的消息为字节, 可以将 java 对象序列化后作为消息体发送。

总结

以上就是这篇文章的全部内容了,希望本文的内容对大家的学习或者工作具有一定的参考学习价值,如果有疑问大家可以留言交流,谢谢大家对服务器之家的支持。

原文链接:https://www.cnblogs.com/Finley/p/10126315.html

延伸 · 阅读

精彩推荐
  • Java教程详解Mybatis通用Mapper介绍与使用

    详解Mybatis通用Mapper介绍与使用

    目前通用mapper只支持对单表的操作,对单表的增删改查,无需在mapper.xml写对应的sql语句,只需要我们调用相应的接口,对于快速开发极为方便,感兴趣的小...

    Cipher''s Blog3902021-05-07
  • Java教程Eclipse智能提示及快捷键

    Eclipse智能提示及快捷键

    本文主要介绍了Eclipse智能提示及快捷键的相关知识,具有很好的参考价值。下面跟着小编一起来看下吧...

    唯一方向1532020-08-26
  • Java教程Java基础教程_判断语句if else

    Java基础教程_判断语句if else

    下面小编就为大家带来一篇Java基础教程_判断语句if else。小编觉得挺不错的,现在就分享给大家,也给大家做个参考。一起跟随小编过来看看吧 ...

    jingxian4502020-05-12
  • Java教程java生成图片验证码示例代码

    java生成图片验证码示例代码

    这篇文章主要为大家详细介绍了java生成图片验证码示例代码,文中示例代码介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们可以参考一下 ...

    奔跑的四块五1952020-06-05
  • Java教程RSA加密算法java简单实现方法(必看)

    RSA加密算法java简单实现方法(必看)

    下面小编就为大家带来一篇RSA加密算法java简单实现方法(必看)。小编觉得挺不错的,现在就分享给大家,也给大家做个参考。一起跟随小编过来看看吧 ...

    haole196762020-06-16
  • Java教程详解Kotlin中如何实现类似Java或C#中的静态方法

    详解Kotlin中如何实现类似Java或C#中的静态方法

    Kotlin中如何实现类似Java或C#中的静态方法,本文总结了几种方法,分别是:包级函数、伴生对象、扩展函数和对象声明。这需要大家根据不同的情况进行选...

    嘎子3932020-11-02
  • Java教程Java中Thread.join()的使用方法

    Java中Thread.join()的使用方法

    这篇文章主要介绍了Java中Thread.join()的使用方法,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面...

    huangzejun1542020-08-01
  • Java教程Java线程池使用与原理详解

    Java线程池使用与原理详解

    这篇文章主要为大家详细介绍了Java线程池使用与原理的相关资料,具有一定的参考价值,感兴趣的小伙伴们可以参考一下...

    _Yasin8192021-01-15