- ubuntu12.04环境下使用kvm ioctl接口实现最简单的虚拟机
- Ubuntu 通过无线网络安装Ubuntu Server启动系统后连接无线网络的方法
- 在Ubuntu上搭建网桥的方法
- ubuntu 虚拟机上网方式及相关配置详解
CFSDN坚持开源创造价值,我们致力于搭建一个资源共享平台,让每一个IT人在这里找到属于你的精彩世界.
这篇CFSDN的博客文章RabbitMQ消息中间件示例详解由作者收集整理,如果你对这篇文章有兴趣,记得点赞哟.
前言 。
rabbitmq 是使用 erlang 语言开发的消息中间件, 其遵循了高级消息队列协议(advanced message queuing protocol, amqp).
与 kafka 等消息队列相比,rabbitmq 最大的优势在于其较高的可靠性
因为具有较高可靠性和一致性, rabbitmq 可以胜任订单处理、秒杀等一致性要求较高的业务场景.
rabbitmq 概念与机制 。
rabbitmq 中的概念模型
交换机(exchange) 。
生产者发送的消息会首先送到交换机(exchange), 交换机根据自身类型和消息的 routing-key 等信息将消息投递到绑定的消息队列中.
rabbitmq中的四种标准交换机
direct: 如果消息的 routing-key 与队列的 binding-key 完全相同,direct类型的交换机则会将消息投递到该队列中.
topic: 允许队列的 binding-key 中包含通配符*和#, topic 交换机会将消息投递到 binding-key 与 routing-key 匹配的队列中.
fanout: fanout 交换机不进行任何匹配, 将消息投递到所有绑定的队列 。
header: header 交换机根据消息头进行投递,现在已较少使用 。
我们可以使用 rabbitmq 的插件机制使用第三方交换机或自行开发交换机。如实现延时投递的.
消息头中的delivery-mode可以设置为 persistent(持久化) 或者 transient(易失)。 exchange 和 queue 在处理持久化的消息时都会先将消息写入磁盘中再进行下一步处理, 即使 rabbitmq 崩溃也不会丢失.
消费者客户端通常使用的channel.basicconsume使用推(push)模式投递消息, 即当有新消息时 broker 通过 channel 主动向客户端发送消息。客户端也可以使用channel.basicget从 broker 拉取消息.
ack机制 。
rabbitmq 提供了确认送达(acknowledge)机制保证消息被正确处理不会丢失.
确认送达的回执有三种
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 。
最后此篇关于RabbitMQ消息中间件示例详解的文章就讲到这里了,如果你想了解更多关于RabbitMQ消息中间件示例详解的内容请搜索CFSDN的文章或继续浏览相关文章,希望大家以后支持我的博客! 。
我想读取 RabbitMQ 队列中未确认消息的负载或 messageId。这可能吗? 我想这样做的原因是我尝试使用 RabbitMQ 死信功能来构建一个循环以定期自动生成消息。简而言之,创建两个队列
除了 vFabric 由 VMWare 提供商业支持之外,vFabric RabbitMQ 和 RabbitMQ 之间的主要区别是什么? 最佳答案 来自 source : We also produc
RabbitMQ 集群中有如下三个节点。 在 RabbitMQ 中,有两个队列,q1 和 q2。 q1 和q2 的主副本分布在不同的节点上。两个队列都被其他节点镜像。 三个节点前面有一个负载均衡器。
我希望在谷歌计算引擎上实现 rabbitmq 来处理我的 android 和 ios 消息传递应用程序上的消息。我听说 rabbitmq 可能非常耗电,所以我想知道解决这个问题的最佳解决方案是什么?我
是否可以在 RabbitMQ 服务器(管理插件 View )中查看连接单元的主机名,而不仅仅是 IP/端口?我们使用动态连接位置,这样更容易识别客户...... 最佳答案 不,没有这样的内置功能。 作
我正在阅读 RabbitMQ in Action 书,仍在第 2 章中,但作者说的一件事让我感到困惑。您设置了一个交换并发送了一条消息,两个订阅者正在监听队列。当第一条消息进来时,第一个订阅者得到它,
我正在使用 RabbitMQ 将所有消息排队,并将消息作为 SMS 发送给各个消费者。我正在使用直接交换,并且我已经正确地创建了一个到带有路由键的队列的绑定(bind)。问题是,当我尝试发布消息时,我
我们正在使用微服务架构在 nodejs 中实现 Web-API。每个服务都会公开 HTTP 端点,以便应用程序/网站可以与其交互。为了同步不同的数据库,我们目前使用 RabbitMQ。微服务可以在扇出
我计划在 RabbitMQ 消息头中存储堆栈跟踪。消息 header 是否有大小限制? 最佳答案 RabbitMQ 默认使用 AMQP 版本 0.9.1。根据AMQP protocol specifi
无法理解 exclusive queue 和 exclusive consumer 之间的区别,想知道我是否理解正确。 假设我有一个 queue、consumer1 和 consumer2。 我的理解
发布到 RabbitMQ 队列(发布/订阅模型)时消息的最大大小是多少? 我在文档中看不到任何明确的限制,但我认为有一些指导方针。 提前致谢。 最佳答案 我在做比较亚马逊队列服务和 RabbitMQ
我可以使用 Publish/Subscribe 创建扇出交换RabbitMQ Java 教程,任何连接的消费者都会收到一条消息的副本。我不想以动态/编程方式声明交换和绑定(bind),而是想在连接任何
java的 native rabbitmq客户端允许在连接设置上设置心跳,例如: import com.rabbitmq.client.ConnectionFactory; ... Connectio
我开始着手一个新项目,我们被要求将系统构建为一系列微服务,使用 RabbitMQ 作为它们之间的通信层。 在开发 REST API 时,我倾向于使用接受 HTTP header 来控制版本控制,我看到
在 Rabbit MQ 中使用集群时,我计划使用竞争订阅者模式。 Producer : 1 Exchange : 1 direct Queue : 1 Consumers : n (multiple)
是否可以实现 aggregator pattern在 RabbitMQ 中? 我有 A … N在发送到另一个队列之前我需要等待/聚合的消息 X . 所以我想我会有一些 唯一 ID 确保消息被路由 独家
我正在使用RabbitMQ向用户发送通知。用户可以随时读取其队列。 我面临的问题是,队列在夜间充满了很多通知,而当用户在早上返回时,他必须顺序处理这些消息。这些通知中有很多甚至是重复的。 我认为在发布
是否可以延迟通过 RabbitMQ 发送消息? 例如,我想在 30 分钟后使客户端 session 过期,并且我发送了一条将在 30 分钟后处理的消息。 最佳答案 您可以尝试两种方法: 旧方法:在每个
RabbitMQ 在单个服务器上可以处理的最大队列数是多少? 这取决于内存吗?它取决于 erlang 进程吗? 最佳答案 RabbitMQ 代理内部没有任何硬编码限制。代理将利用所有可用资源(除非您对
RabbitMQ Server 使用或需要在防火墙上为节点集群打开哪些端口? 我的 /usr/lib/rabbitmq/bin/rabbitmq-env 设置如下,我假设需要 (35197)。 SER
我是一名优秀的程序员,十分优秀!