ActiveMQ处理MapMessage队列

港控/mmm° 2023-02-12 09:28 0阅读 0赞

### 控制层 ###

//新增队列（发送MapMessage消息）
    @RequestMapping("/sendMapMessage")
    @ResponseBody
    public String sendMapMessage(@RequestBody String data) {
    	// 字符串转json
    	Map<String, Object> map = (Map<String, Object>) JSONObject.parse(data);
    	String connectionFactory = map.get("connectionFactory") + "";
    	Map<String, String> content = (Map<String, String>) map.get("content");
    	String queueName = map.get("queueName")+"";
    	String msg=null;
    	try {
    		msg = mQQueueService.sendMapMessage(connectionFactory, queueName,content);
    	} catch (JMSException e) {
    		// TODO Auto-generated catch block
    		e.printStackTrace();
    	}
    	return msg;
    }
    	
    //消费队列（获取MapMessage消息）
    @RequestMapping("/achieveMapMessage")
    @ResponseBody
    public Map<String, Object> achieveMapMessage(@RequestBody String data) {
    	// 字符串转json
    	Map<String, Object> map = (Map<String, Object>) JSONObject.parse(data);
    	String connectionFactory = map.get("connectionFactory") + "";
    	String studentNo = map.get("studentNo")+"";
    	String queueName = map.get("queueName")+"";
    	Map<String, Object> mapMapMessage=null;
    	try {
    		 mapMapMessage= mQQueueService.achieveMapMessage(connectionFactory, queueName,studentNo);
    	} catch (JMSException e) {
    		e.printStackTrace();
    	}
    	return mapMapMessage;
    }

### 业务层 ###

// 发送MapMessage演示
    @Override
    public String sendMapMessage(String connectionFactory, String queueName, Map<String, String> content) throws JMSException {
        ConnectionFactory cf = new ActiveMQConnectionFactory(connectionFactory);
    	Connection connection = cf.createConnection();
    	connection.start(); //
    	// 创建会话（需要创建事务，消息自动确认）
    	Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
    	// 循环创建队列
    	Destination destination = session.createQueue(queueName);
    	// 创建生产者，即消息发送方。
    	MessageProducer producer = session.createProducer(destination);
    	// 这里以一个学号对应一个学生为例
    	// key表示学号，key对应的value表示学生姓名
    	// 假如设置属性是班级编号1203
    	for (String key : content.keySet()) {
    		String value = content.get(key);
    		MapMessage mapMessage = session.createMapMessage();
    		mapMessage.setStringProperty("classNo", "1203");
    		mapMessage.setString(key, value);
    		producer.send(mapMessage);
    	}
    	try {
    		session.commit();
    	} finally {
    		session.close();
    		connection.close();
    	}
    	return "SUCCESS";
    }
    
    // 获取MapMessage演示
    @Override
    public Map<String, Object> achieveMapMessage(String connectionFactory, String queueName, String studentNo) throws JMSException {
    	ConnectionFactory cf = new ActiveMQConnectionFactory(connectionFactory);
    	Connection connection = cf.createConnection();
    	connection.start();
    	// 确认消息被接受
    	final Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
    	Destination destination = session.createQueue(queueName);
    	MessageConsumer consumer = session.createConsumer(destination);
    	MapMessage mm = (MapMessage) consumer.receive();
    	// 根据学号获取学生姓名
    	String value = mm.getString(studentNo);
    	// 根据学号获取班级编号
    	String property = mm.getStringProperty("classNo");
    	System.out.println("学号：" + studentNo);
    	System.out.println("姓名：" + value);
    	System.out.println("班级：" + property);
    	Map<String, Object> hashMap = new HashMap<String, Object>();
    	// 如果根据学号可以查询到学生姓名，则消费队列。
    	if (value == null) {
    		hashMap.put("msg", "查询该学号失败。");
    	} else {
    		session.commit();
    		hashMap.put("msg", "查询该学号成功。");
    	}
    	hashMap.put("studentNo", studentNo);
    	hashMap.put("studentName", value);
    	hashMap.put("classNo", property);
    	try {
    		return hashMap;
    	} finally {
    		session.close();
    		connection.close();
    	}
    }

### 新增队列 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NzA2NTcw_size_16_color_FFFFFF_t_70][]

### 消费队列 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NzA2NTcw_size_16_color_FFFFFF_t_70 1][]

### 启动ActiveMQ后台 ###

//初始账号：admin
    //初始密码：admin
    http://192.168.0.113:8161/admin

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NzA2NTcw_size_16_color_FFFFFF_t_70 2][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NzA2NTcw_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200525173158878.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NzA2NTcw,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NzA2NTcw_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200525173244258.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NzA2NTcw,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NzA2NTcw_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200525173356146.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5NzA2NTcw,size_16,color_FFFFFF,t_70