热门标签 | HotTags
当前位置:  开发笔记 > 编程语言 > 正文

RabbitMQ运用负载均衡与消息持久化的实现

 Rabbitmq是对AMQP协议的一种实现。使用范围也比较广泛,主要用于消息异步通讯。

 

Rabbitmq 是对AMQP协议的一种实现。使用范围也比较广泛,主要用于消息异步通讯。



一,默认情况下Rabbitmq使用轮询(round-robin)方式转发消息。为了较好实现负载,可以在消息接收方指定,每次接收到一条,这样可以缓解单一服务器压力。

代码如下:



ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
cOnnection= factory.newConnection();
final Channel channel = connection.createChannel();
channel.basicQos(1);//设置每次接收一条
为了保证消息不丢失,取消自动ACK,改为只有在完全处理消息后再ACK。
如:

Consumer cOnsumer= new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag,
Envelope envelope,
AMQP.BasicProperties properties,
byte[] body)
throws IOException {
String message = new String(body, "UTF-8");
try {
Thread.sleep(10000);
}
catch(Exception ex)
{
}
System.out.println("received Message:" + message);
channel.basicAck(envelope.getDeliveryTag(), false);//处理完成后ACK
}
};
channel.basicConsume(QUEUE_NAME, false, consumer);//取消自动ACK
二,为了保证在Rabbitmq在宕机后,仍不丢失消息,需要将队列和发布的消息都声明为可持久化的。
如:

channel.queueDeclare(QUEUE_NAME, true, false, false, null);

channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, bytes);

三,Rabbitmq 的MessageModel(消息模型)
在Rabbitmq的消息模型中,我们决不应该将消息直接发送到queue.事实上,消息发送者并不关心消息是否被路由或被入队列或被接收并处理。
生产者应只与Exchange(交换器)打交道,
Exchange的作用:从生产者接受消息,向消费者发送消息。
The exchange must know exactly what to do with a message it receives. Should it be appended to a particular queue? Should it be appended to many queues? Or should it get discarded. The rules for that are defined by the exchange type.【所以交换器必须准确地知道怎样处理消息,是否应该加到一个指定的队列,还是发送到多个队列,还是应该抛弃该消息。】
指定Exchange的Rule,即以何种方式转发消息。
Rabbitmq共有四种:direct,topic ,headers和 fanout,NamelessExchange.

fanout,这种方式很简单,就是一个广播,把消息转给所有的订阅者;有几个订阅者,消息就会被复制几份。

NamelessExchange,无Exchange,消息以轮询(round-robin)方式,发送给消费者,通过routingKey识别对应的消费者。

【提示】:

rabbitmqctl list_exchanges ,用于查看当前Rabbitmq正在运行的交换器;rabbitmqctl list_bindings,查看当前绑定数
eg:
生产者只负责发送消息,而不关心这些消息是否被处理,也不关心消息是否被抛弃;消息由Exchange根据具体rule处理。

private static final String EXCHANGE_NAME = "logs";
private static final String EXCHANGE_TYPE="fanout";
public static void main(String[] argv)
throws java.io.IOException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection cOnnection= factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME,EXCHANGE_TYPE);
String message = getMessage(argv);
channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");
channel.close();
connection.close();
}

 如:以fanout方式处理消息:消息会发送给所有的订阅者,(与routingKey无关)

private static final String exchangeName="logs";
private static final String exchangeType = "fanout";
public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {
Connection cOnnection= null;
try {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
cOnnection= factory.newConnection();
final Channel channel = connection.createChannel();
int prefetchCount = 1;
channel.basicQos(prefetchCount);
channel.exchangeDeclare(exchangeName,exchangeType);
//创建一个队列,用于接收消息
String queueName= channel.queueDeclare().getQueue();
channel.queueBind(queueName,exchangeName,"");
System.out.println("Waiting for messages...,over it ,Press CTRL+C ");
Consumer cOnsumer= new DefaultConsumer(channel) {
public void handleDelivery(String consumerTag,
Envelope envelope,
AMQP.BasicProperties properties,
byte[] body)
throws IOException {
String message = new String(body, "UTF-8");
System.out.println("received Message:" + message);
try {
doWork(message);
} catch (InterruptedException e) {
e.printStackTrace();
}
finally {
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
channel.basicConsume(queueName, false, consumer);
} finally {
}
}

Bindings,将队列绑定到Exchange,说明可以从Exchange接收消息。【A binding is a relationship between an exchange and a queue. This can be simply read as: the queue is interested in messages from this exchange.】

channel.queueBind(queueName, EXCHANGE_NAME, "routingKey");
direct,只有当routingKey,与bindingKey相同Exchange才能会推送消息。
如:
生产者:

channel.basicPublish(exchangeName, "routingKey", MessageProperties.PERSISTENT_TEXT_PLAIN, bytes);//片断

生产者在发送消息时指定routingKey.
消费者:可以绑定多个Key,以接收来自多个routingKey的消息

for (String bindingKey : typeArr) {
channel.queueBind(queueName, exchangeName, bindingKey);
}
//接收者绑定Key.
topic ,生产者在发送消息时,指定准确的routingKey(多个单词以.号分隔),当bindingKey模式匹配到routingKey时,则接收消息。
注意:routingKey和bindingKey的长度不能超过255.
*(star),只匹配一个单词。#(hash),能匹配0个或多个单词
三RPC
分布式远程调用。

 

 



   



推荐阅读
  • RabbitMQ的消息持久化处理
    1、RabbitMQ的消息持久化处理,消息的可靠性是RabbitMQ的一大特色,那么RabbitMQ是如何保证消息可靠性的呢——消息持久化。2、auto ... [详细]
  • http:my.oschina.netleejun2005blog136820刚看到群里又有同学在说HTTP协议下的Get请求参数长度是有大小限制的,最大不能超过XX ... [详细]
  • 消息中间件RabbitMQ 高级特性之消费端ACK与重回队列
    什么是消费端的ACK和重回队列?消费端的手工ACK和NACK消费端进行消费的时候,如果由于业务异常我们可以进行日志的记录,然后进行补偿如果由于服务器宕机等严重问题 ... [详细]
  • 本文整理了Java中org.apache.activemq.util.ByteArrayInputStream.<init>()方法的一些代码示例,展示了 ... [详细]
  • 讨伐Java多线程与高并发——MQ篇
    本文是学习Java多线程与高并发知识时做的笔记。这部分内容比较多,按照内容分为5个部分:多线程基础篇JUC篇同步容器和并发容器篇线程池篇MQ篇本篇 ... [详细]
  • 本文介绍了C#中生成随机数的三种方法,并分析了其中存在的问题。首先介绍了使用Random类生成随机数的默认方法,但在高并发情况下可能会出现重复的情况。接着通过循环生成了一系列随机数,进一步突显了这个问题。文章指出,随机数生成在任何编程语言中都是必备的功能,但Random类生成的随机数并不可靠。最后,提出了需要寻找其他可靠的随机数生成方法的建议。 ... [详细]
  • 本文详细介绍了Spring的JdbcTemplate的使用方法,包括执行存储过程、存储函数的call()方法,执行任何SQL语句的execute()方法,单个更新和批量更新的update()和batchUpdate()方法,以及单查和列表查询的query()和queryForXXX()方法。提供了经过测试的API供使用。 ... [详细]
  • CF:3D City Model(小思维)问题解析和代码实现
    本文通过解析CF:3D City Model问题,介绍了问题的背景和要求,并给出了相应的代码实现。该问题涉及到在一个矩形的网格上建造城市的情景,每个网格单元可以作为建筑的基础,建筑由多个立方体叠加而成。文章详细讲解了问题的解决思路,并给出了相应的代码实现供读者参考。 ... [详细]
  • [大整数乘法] java代码实现
    本文介绍了使用java代码实现大整数乘法的过程,同时也涉及到大整数加法和大整数减法的计算方法。通过分治算法来提高计算效率,并对算法的时间复杂度进行了研究。详细代码实现请参考文章链接。 ... [详细]
  • 前景:当UI一个查询条件为多项选择,或录入多个条件的时候,比如查询所有名称里面包含以下动态条件,需要模糊查询里面每一项时比如是这样一个数组条件:newstring[]{兴业银行, ... [详细]
  • Java SE从入门到放弃(三)的逻辑运算符详解
    本文详细介绍了Java SE中的逻辑运算符,包括逻辑运算符的操作和运算结果,以及与运算符的不同之处。通过代码演示,展示了逻辑运算符的使用方法和注意事项。文章以Java SE从入门到放弃(三)为背景,对逻辑运算符进行了深入的解析。 ... [详细]
  • 抖音服务器带宽有多大,才能供上亿人同时刷?
    最近看到一个有意思的提问:抖音服务器带宽有多大,为什么能够供那么多人同时刷?今天来给大家科普一下。 ... [详细]
  • 1、概述首先和大家一起回顾一下Java消息服务,在我之前的博客《Java消息队列-JMS概述》中,我为大家分析了:然后在另一篇博客《Java消息队列-ActiveMq实战》中 ... [详细]
  • 域名解析系统DNS
    文章目录前言一、域名系统概述二、因特网的域名结构三、域名服务器1.根域名服务器2.顶级域名服务器(TLD,top-leveldomain)3.权威(Authoritative)域名 ... [详细]
  • Docker安装Rabbitmq(配合宝塔)
    篇首语:本文由编程笔记#小编为大家整理,主要介绍了Docker安装Rabbitmq(配合宝塔)相关的知识,希望对你有一定的参考价值。一、事前准备 ... [详细]
author-avatar
张伊韵育财育信
这个家伙很懒,什么也没留下!
PHP1.CN | 中国最专业的PHP中文社区 | DevBox开发工具箱 | json解析格式化 |PHP资讯 | PHP教程 | 数据库技术 | 服务器技术 | 前端开发技术 | PHP框架 | 开发工具 | 在线工具
Copyright © 1998 - 2020 PHP1.CN. All Rights Reserved | 京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有