逍遥bxf飞雪
- 浏览: 21148 次
-
最新评论
RabbitMQ
1.消费者端
package com.guoxin.rabbitmq;
import java.io.IOException;
import com.guoxin.parser.download.CreateLog;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.AMQP.Queue.DeclareOk;
public class Consumer1 {
private final String EXCHANGE_NAME = "direct_logs";
private final String QUEUE_NAME = "StaticQueue";
boolean flag = true;
// 创建链接工厂
private static ConnectionFactory factory;
// 创建链接
Connection connection = null;
Channel channel = null;
String task = "";
private static int a=0;
public String consumer() {
try {
factory=Consumer1.getConnectionFactory();
factory.setRequestedChannelMax(20);
factory.setRequestedFrameMax(0);
// factory=new ConnectionFactory();
factory.setHost("192.168.1.22");
connection = factory.newConnection();
// 创建信息通道
channel = connection.createChannel();
try {
// 声明Exchange非持久化
channel.exchangeDeclare(EXCHANGE_NAME, "direct", true);
// 参数说明：第�?��参数为队列的名字。第二个参数为：设置为true，则rabbitmq在重启之后会重新建立
// 第三个参数：设置为true，标识只有创建了该队列的消费者才可以去连接该队列
// 第四个参数：设置为false为最后一个消费�?断开后不会自动删除�?
// 第五个参数为：队列的其他参数，没有就可以设置为空
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "StaticState");
DeclareOk a = channel.queueDeclarePassive(QUEUE_NAME);
//获取队列中的消息个数
int c = a.getMessageCount();
// System.out.println(c);
// System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
// 声明一个消费者
QueueingConsumer consumer = new QueueingConsumer(channel);
// 新消息到达的回调
channel.basicConsume(QUEUE_NAME, false, consumer);
channel.basicQos(1);
while (flag) {
// QueueingConsumer.Delivery封装一个消息，和bean的结构相同
QueueingConsumer.Delivery delivery = consumer
.nextDelivery();
// 获取消息
String message = new String(delivery.getBody());
task = message;
channel.basicAck(delivery.getEnvelope().getDeliveryTag(),
false);
// System.out.println(task);
break;
}
} catch (Exception e) {
e.printStackTrace(); }
} catch (IOException e1) {
e.printStackTrace(); } finally {
try {
channel.close();
connection.close();
} catch (IOException e) {
e.printStackTrace(); }
}
return task;
}
public synchronized static ConnectionFactory getConnectionFactory(){
if(factory==null){
a++;
System.out.println("获取次数:"+a);
factory=new ConnectionFactory();
}
return factory;
}
public static void main(String[] argv) throws Exception {
Consumer1 consumer1 = new Consumer1();
consumer1.consumer();
}
}
2.生产者端
package com.guoxin.util;
import java.io.IOException;
import java.util.List;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.springframework.scheduling.quartz.QuartzJobBean;
import com.guoxin.bean.Block;
import com.guoxin.common.util.Config;
import com.guoxin.common.webservice.BmsMailSend;
import com.guoxin.log.CreateLog;
import com.guoxin.service.Iservice.GetBlockUrlService;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;
import com.rabbitmq.client.AMQP.Queue.DeclareOk;
import com.sun.codemodel.JNullType;
@SuppressWarnings("deprecation")
public class DMSMethod extends QuartzJobBean{
private static final String EXCHANGE_NAME = "direct_logs";
private final String QUEUE_NAME = "StaticQueue";
private GetBlockUrlService blockservice;
public DMSMethod(){
blockservice = (GetBlockUrlService)MyApplicationContextUtil.getContext().getBean("blockService");
}
protected void executeInternal(JobExecutionContext arg0)
throws JobExecutionException {
ConnectionFactory factory;
Connection connection = null;
Channel channel = null;
try {
factory= new ConnectionFactory();
factory.setHost("192.168.1.22");
// factory.setRequestedChannelMax(20);
// factory.setRequestedFrameMax(0);
connection= factory.newConnection();
channel=connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "direct",true);
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
DeclareOk a = channel.queueDeclarePassive(QUEUE_NAME);
int c = a.getMessageCount();
//获得静态版块的链接
List<Block> staticList=blockservice.queryBlockUrlStatic();
int abc=0;
for(Block block:staticList){
String task=block.getBlock_url();
if(c<15){
channel.basicPublish(EXCHANGE_NAME,"StaticState", MessageProperties.PERSISTENT_TEXT_PLAIN, task.getBytes());
System.out.println("队列中剩余:"+c);
}
}
} catch (Exception e) {
BmsMailSend.sendMail(Short.valueOf(Config.getAttribute("systemNo")),"亲,又发生异常了!" ,(new Throwable().getStackTrace()[0]).toString()+ e.toString());
CreateLog.getCreateLogInstance("errorLog").CreateLogs(e,
"com.guoxin.util.DMSMethod");
}finally{
try {
channel.close();
connection.close();
} catch (IOException e) {
System.out.println("异常发生");
BmsMailSend.sendMail(Short.valueOf(Config.getAttribute("systemNo")),"亲,又发生异常了!" ,(new Throwable().getStackTrace()[0]).toString()+ e.toString());
CreateLog.getCreateLogInstance("errorLog").CreateLogs(e,
"com.guoxin.util.DMSMethod");
}
}
}
public GetBlockUrlService getBlockservice() {
return blockservice;
}
}
以上是在项目中的实际应用，DMSMethod类是采用了spring的定时器来加载的，定时向队列中插入数据，当队列中的数据个数小于15就往队列中放入新的数据。同时查询是调用的service层的接口。在消费者端，我采用的是将一个链接取出来然后就跳出。一般是消费者端是一个死循环，时时刻刻接收生产者往队列中放入得数据。
2.以下是几个不同交换机类型的小例子
1).第一种:发布式的，此种类型交换机忽略路由键值，会将消息发送到它所能知道的队列中
package com.mq.PublishSubscribe;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
/**
* 把消息分发给所有的消费者fanout交换器很简单，你可能从名字上就能猜测出来，它把消息发送给它所知道的所有队列。这正是我们的日志系统所需要的。
* @author Administrator
*
*/
public class EmitLog {
private static final String EXCHANGE_NAME = "logs";
public static void main(String[] argv) throws Exception {
//创建链接工厂
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.1.22");
//创建链接
Connection connection = factory.newConnection();
//创建信息通道
Channel channel = connection.createChannel();
//声明Exchange非持久化
//channel.exchangeDeclare(EXCHANGE_NAME, "fanout",true);这是一个持久化的交换机
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
//消息
String message = getMessage(argv);
String message1="删除";
String message2="shanchu1";
//发送消息，这个发送消息是忽略路由键值的。它会将消息发送给所有与这个交换机绑定的队列上
channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
channel.basicPublish(EXCHANGE_NAME, "", null, message1.getBytes());
channel.basicPublish(EXCHANGE_NAME, "", null, message2.getBytes());
System.out.println(" [x] Sent '" + message + "'");
channel.close();
connection.close();
}
private static String getMessage(String[] strings){
if (strings.length < 1)
return "info: Hello World!";
return joinStrings(strings, " ");
}
private static String joinStrings(String[] strings, String delimiter) {
int length = strings.length;
if (length == 0) return "";
StringBuilder words = new StringBuilder(strings[0]);
for (int i = 1; i < length; i++) {
words.append(delimiter).append(strings[i]);
}
return words.toString();
}
}
package com.mq.PublishSubscribe;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.AMQP.Queue.DeclareOk;
import com.rabbitmq.client.AMQP.Queue.DeleteOk;
public class ReceiveLogs {
private static final String EXCHANGE_NAME = "logs";
public static void main(String[] argv) throws Exception {
//创建链接工厂
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.1.22");
//创建链接
Connection connection = factory.newConnection();
//创建信息通道
Channel channel = connection.createChannel();
//声明Exchange非持久化
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
//采用系统的自动声明的消息队列名
String queueName = channel.queueDeclare().getQueue();
//将队列绑定到交换机上，没有设定路由键值的，也就是说与这个交换机绑定的所有的队列都会接收到交换机中的消息
channel.queueBind(queueName, EXCHANGE_NAME, "");
// 参数说明：第一个参数为队列的名字。第二个参数为：设置为true，则rabbitmq在重启之后会重新建立
//第三个参数：设置为true，标识只有创建了该队列的消费者才可以去连接该队列
//第四个参数：设置为false为最后一个消费者断开后不会自动删除。
//第五个参数为：队列的其他参数，没有就可以设置为空
channel.queueDeclare("queue123", true, false, false, null);
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
//声明一个消费者
QueueingConsumer consumer = new QueueingConsumer(channel);
//新消息到达的回调
channel.basicConsume(queueName, true, consumer);
while (true) {
//QueueingConsumer.Delivery封装一个消息，和bean的结构相同
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
//获取消息
String message = new String(delivery.getBody());
//获取消息信封
Envelope message1 =delivery.getEnvelope();
//获取消息属性
AMQP.BasicProperties message2 =delivery.getProperties();
System.out.println(" [x] Received '" + message + "'");
System.out.println(" [x] Received Envelope'" + message1 + "'");
System.out.println(" [x] Received AMQP.BasicProperties'" + message2 + "'");
}
}
}
2).特定类型的direct
package com.mq.Routing;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
public class EmitLogDirect {
private static final String EXCHANGE_NAME = "direct_logs";
public static void main(String[] argv) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.1.22");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "direct");
String severity = getSeverity(argv);
String message = getMessage(argv);
channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());
System.out.println(" [x] Sent '" + severity + "':'" + message + "'");
channel.close();
connection.close();
}
private static String getSeverity(String[] strings){
if (strings.length < 1)
return "info";
return strings[0];
}
private static String getMessage(String[] strings){
if (strings.length < 2)
return "Hello World!";
return joinStrings(strings, " ", 1);
}
private static String joinStrings(String[] strings, String delimiter, int startIndex) {
int length = strings.length;
if (length == 0 ) return "";
if (length < startIndex ) return "";
StringBuilder words = new StringBuilder(strings[startIndex]);
for (int i = startIndex + 1; i < length; i++) {
words.append(delimiter).append(strings[i]);
}
return words.toString();
}
}
package com.mq.Routing;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;
public class ReceiveLogsDirect {
private static final String EXCHANGE_NAME = "direct_logs";
public static void main(String[] argv) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.1.22");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "direct");
String queueName = channel.queueDeclare().getQueue();
String queueName1 = channel.queueDeclare().getQueue();
channel.queueBind(queueName, EXCHANGE_NAME, "info");
channel.queueBind(queueName1, EXCHANGE_NAME, "info1");
// if (argv.length < 1){
// System.err.println("Usage: ReceiveLogsDirect [info] [warning] [error]");
// System.exit(1);
// }
//
// for(String severity : argv){
// channel.queueBind(queueName, EXCHANGE_NAME, severity);
// }
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(queueName, true, consumer);
channel.basicConsume(queueName1, true, consumer);
while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
String routingKey = delivery.getEnvelope().getRoutingKey();
String message1 = new String(delivery.getBody());
String routingKey1 = delivery.getEnvelope().getRoutingKey();
System.out.println(" [x] Received '" + routingKey + "':'" + message + "'");
System.out.println(" [x] Received '" + routingKey1 + "':'" + message1 + "'");
}
}
}
这个例子中由于队列和info绑定的，所以发送过来的消息只能放到队列queueName中。
3).匹配类型的交换机
package com.mq.Topics;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.MessageProperties;
public class EmitLogTopic {
private static final String EXCHANGE_NAME = "topic_logs";
public static void main(String[] argv) {
Connection connection = null;
Channel channel = null;
try {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.1.22");
connection = factory.newConnection();
channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
String routingKey = getRouting(argv);
String message = getMessage(argv);
channel.basicPublish(EXCHANGE_NAME, routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
System.out.println(" [x] Sent '" + routingKey + "':'" + message + "'");
}
catch (Exception e) {
e.printStackTrace();
}
finally {
if (connection != null) {
try {
connection.close();
}
catch (Exception ignore) {}
}
}
}
private static String getRouting(String[] strings){
if (strings.length < 1)
return "anonymous.info.abc";
return strings[0];
}
private static String getMessage(String[] strings){
if (strings.length < 2)
return "Hello World!";
return joinStrings(strings, " ", 1);
}
private static String joinStrings(String[] strings, String delimiter, int startIndex) {
int length = strings.length;
if (length == 0 ) return "";
if (length < startIndex ) return "";
StringBuilder words = new StringBuilder(strings[startIndex]);
for (int i = startIndex + 1; i < length; i++) {
words.append(delimiter).append(strings[i]);
}
return words.toString();
}
}
package com.mq.Topics;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;
public class ReceiveLogsTopic {
private static final String EXCHANGE_NAME = "topic_logs";
private static final String QUEUE_NAME = "DL";
public static void main(String[] argv) {
Connection connection = null;
Channel channel = null;
try {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.1.22");
connection = factory.newConnection();
channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
// String queueName = channel.queueDeclare().getQueue();
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "anonymous.*");
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
//消费者接受到一个消息并作长时间处理时，有可能发生意外状况，
//如运行“Consumer”的机器突然关闭，这时这个消息所要执行的任务
//可能没有得到正确处理。
//RabbitMQ会一直等到处理某个消息的“Consumer”的链接失去之后，
//才确定这个消息没有正确处理，从而RabbitMQ重发这个消息
//初始化“Consumer”时有个auto参数，如果设置为true，这个Consumer在收到消息之后会马上返回ack。
//我们的应用应该是在消息的任务处理完之后再ack，因此初始化“Consumer”时这个参数应该置为false
QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(QUEUE_NAME, false, consumer);
//basicQos方法来限制“Consumer”没有ack的消息数目
channel.basicQos(1);
while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
String routingKey = delivery.getEnvelope().getRoutingKey();
System.out.println(" [x] Received '" + routingKey + "':'" + message + "'");
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}
}
catch (Exception e) {
e.printStackTrace();
}
finally {
if (connection != null) {
try {
connection.close();
}
catch (Exception ignore) {}
}
}
}
}
分享到：
发表评论
相关推荐
`rabbitmq-c`是RabbitMQ的一个C语言客户端库，它使得在C程序中与RabbitMQ服务器进行交互变得更加简单。本文将详细介绍如何使用CMake编译`rabbitmq-c-master`源码，并讨论相关知识点。 首先，我们需要了解CMake，这...
标题 "kettle rabbitmq 插件开发" 涉及的是如何在 Pentaho Kettle（也称为 Spoon）中创建和使用 RabbitMQ 插件。Kettle 是一个开源的数据集成工具，它允许用户进行数据抽取、转换和加载（ETL）操作。RabbitMQ 是一个...
RabbitMQ服务器3.10.5是一款广泛使用的开源消息代理和队列服务器，它基于高级消息队列协议（AMQP）实现。这个版本的RabbitMQ提供了稳定且高效的中间件服务，允许分布式系统中的应用程序进行异步通信，确保数据可靠...
在这个"麒麟v10系统Rabbitmq3.6.10安装包"中，我们将探讨如何在麒麟v10环境下安装和配置RabbitMQ 3.6.10版本。 首先，安装RabbitMQ前需要确保系统满足必要的依赖条件。麒麟v10内核版本为4.19.90-17.ky10.x86_64，这...
RabbitMQ是一个开源的消息代理和队列服务器，广泛用于分布式系统中的消息传递。它基于AMQP（Advanced Message Queuing Protocol）标准，允许应用程序之间异步通信，并提供了高可用性、可扩展性和容错性。RabbitMQ的...
标题中的“flink-sql集成rabbitmq”指的是将Apache Flink的数据流处理能力与RabbitMQ消息队列系统相结合，实现数据的实时处理和传输。Flink是一个开源的流处理框架，提供低延迟、高吞吐量的数据处理，而RabbitMQ是一...
**RabbitMQ实战指南** RabbitMQ是一款广泛应用的开源消息队列系统，它基于Advanced Message Queuing Protocol（AMQP）标准，提供高可用性、可靠性和可扩展性。本实战指南将带你深入理解RabbitMQ的核心概念、安装与...
在本教程中，我们将深入探讨如何在 CentOS 7 操作系统上安装 RabbitMQ，这是一个流行的开源消息代理，基于AMQP（Advanced Message Queuing Protocol）协议。RabbitMQ 使用 Erlang 语言开发，它提供了一个可靠的平台...
RabbitMQ是一款开源的消息队列服务软件，它实现了高级消息队列协议（AMQP），以高性能、健壮和可伸缩性闻名，主要由Erlang语言编写。Erlang是一种适合于构建并发处理能力强、高可用性系统的编程语言，这些特点使得...
【标题】：“TP6使用RabbitMQ” 在PHP框架ThinkPHP6（简称TP6）中集成RabbitMQ是一项常见的任务，用于实现异步处理、消息队列和分布式系统的通信。RabbitMQ是一个开源的消息代理和队列服务器，它遵循AMQP（Advanced...
RabbitMQ是一个开源的消息中间件，它基于Advanced Message Queuing Protocol (AMQP)标准实现，用于在分布式系统中高效地传递消息。RabbitMQ的安装过程在Linux环境下需要依赖于Erlang，而Erlang自身又依赖于某些库，...
《RabbitMQ实战Java版——基于rabbitMQ-demo.zip的详解》 在当今的分布式系统中，消息队列作为异步处理、解耦组件的关键技术，得到了广泛应用。RabbitMQ作为一款开源的消息代理和队列服务器，以其稳定性和易用性...
**RabbitMQ-c源码分析** RabbitMQ-c是一个轻量级且高效的C语言实现的RabbitMQ客户端库。RabbitMQ是一个开源的消息代理和队列服务器，它使用AMQP（Advanced Message Queuing Protocol）协议，广泛应用于分布式系统中...
为了能够有效地监控 RabbitMQ 的性能和状态，Prometheus 提供了一个名为 `rabbitmq_exporter` 的工具。然而，在某些情况下，官方网站可能不直接提供这个插件，这时我们需要从第三方源获取，例如在本例中提到的 `...
在Linux环境中，RabbitMQ是一种广泛使用的开源消息队列系统，它基于AMQP（Advanced Message Queuing Protocol）协议实现。本篇文章将详细讲解如何在Linux上安装RabbitMQ，包括必要的依赖软件Erlang和RabbitMQ服务器...
RabbitMQ是一个基于Erlang语言开发的消息中间件，它遵循AMQP（Advanced Message Queuing Protocol）协议，广泛用于分布式系统中的异步处理和解耦。在这个“RabbitMQ Java测试客户端”项目中，我们可以看到它包含了...
在Java开发中，RabbitMQ是一个非常流行的开源消息队列系统，它基于AMQP（Advanced Message Queuing Protocol）协议，提供了高效、可靠的异步通信能力。在这个场景中，"java rabbitmq动态注册，监听实现"涉及到的主要...
RabbitMQ安装程序Windows版 注意： 以管理员账号进行安装，否则可能会导致无法使用。 安装RabbitMQ之前需要提前安装相对应的Erlang依赖版本，推荐Erlang-25.3版本。 安装路径只能包含 ASCII 字符，强烈建议路径的...
【RabbitMQ性能测试报告】 本测试报告详细记录了对RabbitMQ的性能评估，包括在单机模式和集群模式下的压力和稳定性测试。RabbitMQ是业界广泛使用的开源消息代理，它基于AMQP（Advanced Message Queuing Protocol）...
在本资源中，我们有两个安装包：RabbitMQ 3.12.10和Erlang 26.0.2，这两个是RabbitMQ运行所必需的组件。 Erlang是一种函数式编程语言，以其并发能力、容错性和热代码升级功能而闻名。在RabbitMQ中，Erlang作为基础...