RabbitMQ

程序员文章站 2022-07-12 12:57:58

...

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

rabbitmq可视化工具数据分析（sqlite数据库可视化工具）

利用Python学习RabbitMQ消息队列

spring boot中使用RabbitMQ routing路由详解

spring boot使用RabbitMQ实现topic 主题

Spring Boot整合RabbitMQ开发实战详解

Python_RabbitMQ实现简单的进程间通信

Spring Boot中使用RabbitMQ的示例代码

Python RabbitMQ消息队列实现rpc

RabbitMQ .NET消息队列使用详解

CentOs 7.3中搭建RabbitMQ 3.6单机多实例服务的步骤与使用