`

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) {}
      }
    }
  }
}

 

分享到：
| Spring的定时器
评论
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

    RabbitMQ快速入门及API介绍（401M）

    【课程目录】：---第一章：RabbitMQ介绍----1-什么是消息中间件.mp4----2-RabbitMQ消息队列安装：window环境.mp4----3-RabbitMQ消息队列安装 ：Linux环境.mp4----4-Rabbitmq入口示例：server.mp4----5-rabbitmq入口...

    批量删除rabbitmq的队列或交换机

    采用python编写的批量删除rabbitmq的队列或交换机。 1.修改rabbitmq_delete.py中rabbitmq的配置； 2.执行以下命令： 删除队列: python3 rabbitmq_delete.py -k ‘udata.climb’ -d 1 删除交换机: python3 rabbitmq_...

    rabbitmq-java (2).zip

    rabbitmq

    rabbitmq配置文件 rabbitmq.config

    rabbitmq配置文件，用于rabbitmq管理

    RabbitMQ工具类及测试类(完整版)

    RabbitMQClientUtil是MQ的测试工具类，他封装了fanout、direct、topic三种exchange模式，并包括发送数据和接收数据。...rabbitmq.properties配置文件根据自己需要自行放在，放置完毕后，请自己修改util中的文件地址

    RabbitMQ rabbitmq-server-3.6.12-1.el6.noarch 及其安装所需要的软件打包

    RabbitMQ rabbitmq-server-3.6.12-1.el6.noarch 及其安装所需要的软件打包都在这里面，主要报卡一下软件：socat-1.7.3.2.tar.gz、rabbitmq-server-3.6.12-1.el6.noarch.rpm、rabbitmq-release-signing-key.asc、otp_...

    【BAT必备】rabbitMq面试题

    【BAT必备】rabbitMq面试题【BAT必备】rabbitMq面试题【BAT必备】rabbitMq面试题【BAT必备】rabbitMq面试题【BAT必备】rabbitMq面试题【BAT必备】rabbitMq面试题【BAT必备】rabbitMq面试题【BAT必备】rabbitMq面试题...

    SpringBoot整合RabbitMQ.zip

    SpringBoot整合RabbitMQ的详细过程 **1.该篇博文首先讲述了交换机和队列之间的绑定关系** ①direct、②fanout、③topic **2.然后讲消息的回调** 四种情况下，确认触发哪个回调函数： ①消息推送到server，但是在...

    rabbitmq-server-3.7.3

    RabbitMQ是一个由erlang开发的AMQP（Advanced Message Queue ）的开源实现。AMQP 的出现其实也是应了广大人民群众的需求，虽然在同步消息通讯的世界里有很多公开标准（如 COBAR的 IIOP ，或者是 SOAP 等），但是在...

    windows版本rabbitmq安装包

    windows版本rabbitmq安装包 里面内含rabbitmq-server-3.9.13.exe、otp_win64_24 内含rabbitmq_delayed_message_exchange-3.9.0.ez 插件

    ARM版本的 rabbitmq docker镜像资源

    ARM版本的 rabbitmq 镜像资源 版本 3.8.9 使用拷贝到主机 执行：docker load &lt; rabbitmq_arm3.8.9.tar 生成docker镜像

    rabbitmq 3.9.3 配置文件

    rabbitmq 3.9.3 配置文件

    tp6使用rabbitmq

    tp6使用rabbitmq

    rabbitmqadmin

    RabbitMQ命令行手动创建队列rabbitmqadmin用法 手动创建队列方法：登录http://ip:15672/cli下载 将下载的rabbitmqadmin放到/usr/sbin目录下，并赋予权限：chmod 755 rabbitmqadmin 查看命令帮助信息： python ...

    flink-sql集成rabbitmq

    flink-sql集成rabbitmq

    RabbitMQ消息模式之Confirm确认消息

    理解Confirm消息确认机制 消息的确认，是指生产者投递消息后，如果Broker收到消息，则会给我们生产者一个应答。生产者进行接收应答，用来确定这条消息是否正常的发送到Broker，...import com.rabbitmq.client.Queuein

    rabbitMq window10安装包

    rabbitMq window10安装包

    Linux下安装RabbitMQ

    以rabbitmq3.6版本为例

    RabbitMQ使用指南.pdf

    RabbitMQ下载安装配置使用指南官方手册

    RabbitMQ源码和客户端工具

    RabbitMQ源码和客户端工具RabbitMQ源码和客户端工具

Magicbox
Global site tag (gtag.js) - Google Analytics