Bluemix结合RabbitMq实现消息发送与接收实例

不念不忘少年蓝@ 2021-09-10 19:38 409阅读 0赞

林炳文Evankaka原创作品。转载请注明出处[http://blog.csdn.net/evankaka][http_blog.csdn.net_evankaka]

什么是RabbitMq？

MQ全称为Message Queue, 消息队列（MQ）是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接它们。消 息传递指的是程序之间通过在消息中发送数据进行通信，而不是通过直接调用彼此来通信，直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过 队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。

什么是Bluemix？

BlueMix 是 IBM 基于 Cloud Foundry 的开放云架构实现，使您能够快速创建、部署和管理云应用程序。

本文实现了在Bluemix上绑定rabbitMq,然后在本地运行项目来进行消息的发送和接收。

本文实例访问：[http://javahelloworld2.mybluemix.net/][http_javahelloworld2.mybluemix.net]

本文工程下载：[http://download.csdn.net/detail/evankaka/9427970][http_download.csdn.net_detail_evankaka_9427970]

**BluxMix账号注册：**[https://apps.admin.ibmcloud.com/manage/trial/bluemix.html?cm\_mmc=CMDeveloperGCG-\_-Bluemix-\_-CSDN-\_-onlineeventQ2][https_apps.admin.ibmcloud.com_manage_trial_bluemix.html_cm_mmc_CMDeveloperGCG-_-Bluemix-_-CSDN-_-onlineeventQ2]

# 一、Bluemix项目创建  #

1、新建一个Bluemix的web工程

如何创建工程可看[基于IBM Bluemix部署Java Web项目实战演练][IBM Bluemix_Java Web]

2、添加RabbitMQ服务

![20160217185459307][]

选择CloudMQP

![20160217185514525][]

记下以下数据，工程中会用到

![20160217185602135][]

进入工程，打开如下页面

![20160217185814261][]

此时如果该服务已绑定项目的话，就会弹出如下界面，这个界面说明了如何在客户端来配置RabbitMq的参数，使用客户端能将消息发布的Bluemix这个类似信息中心的地方。如果没有出现此页面，请回去绑定服务到工程！

![20160217185927228][]

# 二、本地运行发布和获取消息 #

本文是基于Maven工程的。所以，先建一个Maven项目工程。最后整个工程目录如下：

![20160219153146191][]

1、添加Jar包依赖

<properties>
    		
    		<spring.version>3.2.8.RELEASE</spring.version>
    		
    		<slf4j.version>1.6.6</slf4j.version>
    		<log4j.version>1.2.12</log4j.version>
    		
    		<junit.version>4.10</junit.version>
    	</properties>
    
    	<dependencies>
    		
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-core</artifactId>
    			<version>${spring.version}</version>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-webmvc</artifactId>
    			<version>${spring.version}</version>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-context</artifactId>
    			<version>${spring.version}</version>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-context-support</artifactId>
    			<version>${spring.version}</version>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-aop</artifactId>
    			<version>${spring.version}</version>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-aspects</artifactId>
    			<version>${spring.version}</version>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-tx</artifactId>
    			<version>${spring.version}</version>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-jdbc</artifactId>
    			<version>${spring.version}</version>
    		</dependency>
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-web</artifactId>
    			<version>${spring.version}</version>
    		</dependency>
    
    		
    		<dependency>
    			<groupId>junit</groupId>
    			<artifactId>junit</artifactId>
    			<version>${junit.version}</version>
    			<scope>test</scope>
    		</dependency>
    
    		
    		
    		<dependency>
    			<groupId>log4j</groupId>
    			<artifactId>log4j</artifactId>
    			<version>${log4j.version}</version>
    		</dependency>
    		<dependency>
    			<groupId>org.slf4j</groupId>
    			<artifactId>slf4j-api</artifactId>
    			<version>${slf4j.version}</version>
    		</dependency>
    		<dependency>
    			<groupId>org.slf4j</groupId>
    			<artifactId>slf4j-log4j12</artifactId>
    			<version>${slf4j.version}</version>
    		</dependency>
    		
    
    		
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-test</artifactId>
    			<version>${spring.version}</version>
    			<scope>test</scope>
    		</dependency>
    
    		
    		<dependency>
    			<groupId>org.springframework.amqp</groupId>
    			<artifactId>spring-rabbit</artifactId>
    			<version>1.4.5.RELEASE</version>
    		</dependency>
    
    
    
    
    
    		<dependency>
    			<groupId>javax.validation</groupId>
    			<artifactId>validation-api</artifactId>
    			<version>1.1.0.Final</version>
    		</dependency>

2、消息发送者

MessageProducer.java内容如下：

package com.lin.producer;
    
    import javax.annotation.Resource;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.stereotype.Service;
    
    /**
     * 功能概要：消息产生,提交到队列中去
     * 
     * @author linbingwen
     * @since  2016年1月15日 
     */
    @Service
    public class MessageProducer {
    	
    	private Logger logger = LoggerFactory.getLogger(MessageProducer.class);
    
    	@Resource
    	private AmqpTemplate amqpTemplate;
    
    	public void sendMessage(Object message){
    	  logger.info("to send message:{}",message);
    	  amqpTemplate.convertAndSend("queueTestKey",message);
    	}
    }

3、消息接收者

MessageConsumer.java内容如下：

package com.lin.consumer;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.core.Message;
    import org.springframework.amqp.core.MessageListener;
    
    /**
     * 功能概要：消费接收
     * 
     * @author linbingwen
     * @since  2016年1月15日 
     */
    public class MessageConsumer implements MessageListener {
    	
    	private Logger logger = LoggerFactory.getLogger(MessageConsumer.class);
    
    	@Override
    	public void onMessage(Message message) {
    		logger.info("receive message:{}",message.toString());
    	}
    
    }

4、定义消息服务中心

这里笔者在rabbitmq.xml将消息生产和消息者都写在一个文件当中，您也可以分开写。配置基本一样。

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
    	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"
    	xsi:schemaLocation="http://www.springframework.org/schema/beans
         http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
         http://www.springframework.org/schema/beans
         http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
         http://www.springframework.org/schema/rabbit
         http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd">
    	
    	<rabbit:connection-factory id="connectionFactory"
    		username="zqwocdmu" password="DRR6tF7DCuEq7ugHwlBP8A9zTwu6jI8D" host = "white-swan.rmq.cloudamqp.com" virtual-host="zqwocdmu" />
    		
    	
    	<rabbit:template id="amqpTemplate"  connection-factory="connectionFactory" 
    	    exchange="exchangeTest" />
    		
    	
    	<rabbit:admin connection-factory="connectionFactory" />
    
    	
    	<rabbit:queue name="queueTest" durable="true" auto-delete="false" exclusive="false" />
    
    	
    	<rabbit:direct-exchange name="exchangeTest" durable="true" auto-delete="false">
    		<rabbit:bindings>
    			<rabbit:binding queue="queueTest" key="queueTestKey"></rabbit:binding>
    		</rabbit:bindings>
    	</rabbit:direct-exchange>
    	
    	
    	<bean id="messageReceiver" class="com.lin.consumer.MessageConsumer"></bean>
    	
    	
        <rabbit:listener-container connection-factory="connectionFactory">
                 <rabbit:listener queues="queueTest" ref="messageReceiver"/>
        </rabbit:listener-container>
    	
    </beans>

其中配置内容和Bluemix上的对应关系如下：

![20160217191151666][]

5、Spring中加载rabbitmq.xml的配置内容

整个application.xml内容如下：

6、测试消息服务

这里笔者简单写了一个不断发送消息和一个不断接收消息的测试

package com.lin;
    
    import javax.annotation.Resource;
    
    import org.junit.Before;
    import org.junit.Test;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.context.ApplicationContext;
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    import org.springframework.retry.backoff.Sleeper;
    
    import com.lin.producer.MessageProducer;
    
    /**
     * 功能概要：
     * 
     * @author linbingwen
     * @since  2016年1月15日 
     */
    public class MessageTest  {
    	
    	private Logger logger = LoggerFactory.getLogger(MessageTest.class);
    	
    	private ApplicationContext context = null;
    	
    	@Before
    	public void setUp() throws Exception {
    	    context = new ClassPathXmlApplicationContext("application.xml");
    	}
    
    	@Test
    	public void should_send_a_amq_message() throws Exception {
           MessageProducer messageProducer = (MessageProducer) context.getBean("messageProducer");
           int a = Integer.MAX_VALUE;
           while (a > 0) {
        	   messageProducer.sendMessage("Hello, I am amq sender num :" + a--);
        	   try {
        		   //暂停一下，好让消息消费者去取消息打印出来
                   Thread.sleep(1000);
               } catch (InterruptedException e) {
                   e.printStackTrace(); 
               }
        
    	   }
    	}
    }

来看看Eclipse上运行的结果：

可以看到，消息不断的产生和消费。

![20160217191354189][]

当然，您也可以进入到RabbitMq的消息服务中心，点击如下：

![20160217191535737][]

点击进去如下显示，这里可以看到消息队列、当前的连接数

![20160217191558202][]

queueTest就是上面的代码中配置的队列

![20160217191639297][]

还可以看到其消息生产和消费的速度等。

![20160217191709235][]

# 三、web项目中来使用rabbitMq #

1、定义一个页面用来发消息和接消息：

<!DOCTYPE html>
    <html>
    <head>
    <meta charset="UTF-8">
    <script src="<%=request.getContextPath()%>/static/js/jquery-2.1.4.min.js"></script>
    </head>
    <body>
    <h2>Hello World!</h2>
    <div>
    <textarea rows="10" cols="40" id = "msg"></textarea>
    </div>
    <div>
    <button id = "sendMsg">发送消息</button>
    <button id = "recvMsg">接收消息</button>
    </div>
    </body>
    <script type="text/javascript">
    $(document).ready(function(){
    	
        //获取当前项目的路径
        var urlRootContext = (function () {
            var strPath = window.document.location.pathname;
            var postPath = strPath.substring(0, strPath.substr(1).indexOf('/') + 1);
            return postPath;
        })();
    	
    	  $("#sendMsg").click(function(){
    		  var msg = $('#msg').val();
    		  if (msg == '') {
    			alert('请输入消息内容 ');
    			return false;
    		  }
    		  
    		  $.ajax( {  
    			    url:urlRootContext + '/sendMsg',// 跳转到 action  
    			    data:{  
    			             message :msg ,  
    			    },  
    			    type:'POST',  
    			    async:false,
    			    dataType:'json',  
    			    success:function(data) {  
    			    	    console.log(data);
    			            alert("发送信息成功！");  
    			            
    			     },  
    			     error: function(xhr, type, exception) {
    			         alert("服务器异常，请稍候重试!");
    			         console.log(xhr);
    			         console.log(type);
    			         console.log(exception);
    			     }
    			});
    	  });
    	  
    	  
    	  $("#recvMsg").click(function(){
    		  $.ajax( {  
    			    url:urlRootContext + '/revcMsg',// 跳转到 action  
    			    type:'POST',  
    			    async:false,
    			    dataType:'json',  
    			    success:function(data) {  
    			    	    console.log(data);
    			            alert("接收一条消息成功！,消息为：" + data.string);  
    			            
    			     },  
    			     error: function(xhr, type, exception) {
    			         alert("服务器异常，请稍候重试!");
    			         console.log(xhr);
    			         console.log(type);
    			         console.log(exception);
    			     }
    			});
    	  });
    	  
    	  
    	});
    
    </script>
    </html>

其中，发消息和接消息都是通过ajax的方式来传递到Controller层中，然后返回json数据到前台。

页面效果如下：

![20160219153300621][]

2、Controller层：

这里需要注意的是消息生产者是的数据服务中心是使用rabbitmq.xml里的配置，但是消息消费都这里是用Java代码来实现的。

package com.lin.controller;
    
    
    import java.io.IOException;
    import java.util.HashMap;
    import java.util.Map;
    
    import javax.annotation.Resource;
    
    import net.sf.json.JSONObject;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.stereotype.Controller;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.ResponseBody;
    
    import com.lin.consumer.MessageConsumer;
    import com.lin.producer.MessageProducer;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.ConsumerCancelledException;
    import com.rabbitmq.client.QueueingConsumer;
    import com.rabbitmq.client.ShutdownSignalException;
    import com.rabbitmq.client.QueueingConsumer.Delivery;
    
    /**
     * 功能概要：UserController
     * 
     * @author linbingwen
     * @since  2015年9月28日 
     */
    @Controller
    public class MessageController  {
    	
    	@Resource
    	private MessageProducer messageProducer;
    	
    	private Logger logger = LoggerFactory.getLogger(getClass());
    	
    	/**
    	 * 显示首页
    	 * @author linbingwen
    	 * @since  2015年10月23日 
    	 * @return
    	 */
    	@RequestMapping("/test")  
    	public String message(){
    		logger.info("open message.jsp");
    		return "message";
    	}
    	
    	/**
    	 * 用来发送消息
    	 * @author linbingwen
    	 * @since  2016年2月19日 
    	 * @param message
    	 * @return
    	 */
    	@RequestMapping("/sendMsg")
    	@ResponseBody
    	public String sendMsg(String message) {
    		logger.info("to send message:{}",message);
    		messageProducer.sendMessage(message);
    		Map<String,Object> map = new HashMap<String,Object>();
    		map.put("flag", true);
    		JSONObject jsonObject =  JSONObject.fromObject(map);
    	    return jsonObject.toString();
    	}
    	
    	
    	/**
    	 * 用来接收消息
    	 * @author linbingwen
    	 * @since  2016年2月19日 
    	 * @param message
    	 * @return
    	 */
    	@RequestMapping("/revcMsg")
    	@ResponseBody
    	public String revcMsg() {
    		String string = getMsg();
    		if (string != null) {
    			
    		} else {
    			string = "null";
    		}
    		Map<String,Object> map = new HashMap<String,Object>();
    		map.put("string", string);
    		JSONObject jsonObject =  JSONObject.fromObject(map);
    	    return jsonObject.toString();
    	}
    	
    	
    	public  String getMsg() {
            ConnectionFactory connFac = new ConnectionFactory() ;
    		
    		connFac.setHost("white-swan.rmq.cloudamqp.com");
    		connFac.setPassword("DRR6tF7DCuEq7ugHwlBP8A9zTwu6jI8D");
    		connFac.setUsername("zqwocdmu");
    		connFac.setVirtualHost("zqwocdmu");
    		Connection conn = null;
    	
    		
    		try {
    			conn = connFac.newConnection();
    		} catch (IOException e) {
    			e.printStackTrace();
    		}
    		
    		Channel channel = null;
    		try {
    			channel = conn.createChannel();
    		} catch (IOException e) {
    			e.printStackTrace();
    		}
    		
    		String queueName = "queueTest";
    		
    		try {
    			channel.queueDeclare(queueName, true, false, false, null) ;
    		} catch (IOException e) {
    			e.printStackTrace();
    		}
    		
    		//配置好获取消息的方式
    		QueueingConsumer consumer = new QueueingConsumer(channel) ;
    		try {
    			channel.basicConsume(queueName, true, consumer) ;
    		} catch (IOException e) {
    			e.printStackTrace();
    		}
    		
    //		//循环获取消息
    //		while(true){
    			
    			//获取消息，如果没有消息，这一步将会一直阻塞
    			Delivery delivery = null;
    			try {
    				delivery = consumer.nextDelivery();
    			} catch (ShutdownSignalException e) {
    				e.printStackTrace();
    			} catch (ConsumerCancelledException e) {
    				e.printStackTrace();
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    			
    			String msg = new String(delivery.getBody()) ;  
    			
    			System.out.println("received message[" + msg + "] from " + queueName);
    //		}
    			try {
    				conn.close();				
    			} catch (IOException e) {
    				e.printStackTrace();
    			}
    	 return  msg;
    	}
    	
    	
    }

3、rabbitMq.xml内容如下：

和上面的工程相比，去掉了消息监听器。而把它放在Controller层中来使用。就是上面的getMsg（）代码。

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
    	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"
    	xsi:schemaLocation="http://www.springframework.org/schema/beans
         http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
         http://www.springframework.org/schema/beans
         http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
         http://www.springframework.org/schema/rabbit
         http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd">
    	
    
    
    <rabbit:connection-factory id="connectionFactory"  
            username="zqwocdmu" password="DRR6tF7DCuEq7ugHwlBP8A9zTwu6jI8D" host = "white-swan.rmq.cloudamqp.com" virtual-host="zqwocdmu" />
    		
    	
    	<rabbit:template id="amqpTemplate"  connection-factory="connectionFactory" 
    	    exchange="exchangeTest" />
    		
    	
    	<rabbit:admin connection-factory="connectionFactory" />
    
    	
    	<rabbit:queue name="queueTest" durable="true" auto-delete="false" exclusive="false" />
    
    	
    	<rabbit:direct-exchange name="exchangeTest" durable="true" auto-delete="false">
    		<rabbit:bindings>
    			<rabbit:binding queue="queueTest" key="queueTestKey"></rabbit:binding>
    		</rabbit:bindings>
    	</rabbit:direct-exchange>
    	
    	
    
    	
    	
    
    
    
    	
    </beans>

这是给远程的Bluemix上绑定的服务中心发消息，

![20160219152713513][]

这是获取消息

![20160219152723604][]

当然，也可以看日志：

![20160219154201469][]

最后一步打war包，然后上传到Bluemix中心即可：

发送消息

![20160219164545993][]

接收消息：

**（这里要注意一定要先发送消息，要不接收消息会一直阻塞会去，整个页面会卡死，而且接收消息的过程非常缓慢，请耐心等待！）**

![20160219164614587][]

本文实例访问：[http://javahelloworld2.mybluemix.net/][http_javahelloworld2.mybluemix.net]

本文工程下载：[http://download.csdn.net/detail/evankaka/9427970][http_download.csdn.net_detail_evankaka_9427970]

[http_blog.csdn.net_evankaka]: http://blog.csdn.net/evankaka
[http_javahelloworld2.mybluemix.net]: http://javahelloworld2.mybluemix.net/
[http_download.csdn.net_detail_evankaka_9427970]: http://download.csdn.net/detail/evankaka/9427970
[https_apps.admin.ibmcloud.com_manage_trial_bluemix.html_cm_mmc_CMDeveloperGCG-_-Bluemix-_-CSDN-_-onlineeventQ2]: https://apps.admin.ibmcloud.com/manage/trial/bluemix.html?cm_mmc=CMDeveloperGCG-_-Bluemix-_-CSDN-_-onlineeventQ2
[IBM Bluemix_Java Web]: http://blog.csdn.net/evankaka/article/details/49836551
[20160217185459307]: /images/20210811/338ed0d92db04ab28acfc3a0921597c3.png
[20160217185514525]: /images/20210811/37526cfb407940a09d8b9edaf0ec526b.png
[20160217185602135]: /images/20210811/2f1e5df50d5e441ba83492b2d4198c53.png
[20160217185814261]: /images/20210811/66019c870a534f4caca3dbdf659f36ab.png
[20160217185927228]: /images/20210811/ec0030d7ca314760b99d04d55192e95c.png
[20160219153146191]: /images/20210811/1284d18f911c4a2c9cc6899ca9f21ced.png
[20160217191151666]: /images/20210811/416c80a2e6054cc9b59e467bd1f8e91d.png
[20160217191354189]: /images/20210811/3ff0943ef9ef4f808757a27a8063bbc5.png
[20160217191535737]: /images/20210811/1efd5012f4df4d53aed0e01592cc37c2.png
[20160217191558202]: /images/20210811/c97da73f16c44c7290d44942bb7b50d3.png
[20160217191639297]: /images/20210811/f8f25b6383d641a7a108bae01bcb698e.png
[20160217191709235]: /images/20210811/2011ff2c380d4c5ead36924b94f00fec.png
[20160219153300621]: /images/20210811/2a0aeef30d5c42d4a7a635d79fb49eb5.png
[20160219152713513]: /images/20210811/573e233039da434fa1b5d554908014c4.png
[20160219152723604]: /images/20210811/7981673e81f648809845ad08c543d5d8.png
[20160219154201469]: /images/20210811/e6861debe1e64dd698db319c076430f4.png
[20160219164545993]: /images/20210811/61ac7280133c480c83ca84ca35df7759.png
[20160219164614587]: /images/20210811/220603e247a54761a3086a58df10d92f.png