Spring - Message Queue 구현

Message Queue

출처 : 소프트웨어 개발자가 바라보는 세상 :: spring에서 ActiveMQ 접근 방법

메시지 객체

package com.iot.mq;

public class MQMessage {

	private String name;
	private String value;

	public void setName(String name) {
		this.name = name;
	}

	public String getName() {
		return name;
	}

	public void setValue(String value) {
		this.value = value;
	}

	public String getValue() {
		return value;
	}

	@Override
	public String toString() {
		return "MQMessage [name=" + name + ", value=" + value + "]";
	}
}

메시지 Sender

package com.iot.mq;

import org.springframework.jms.core.JmsTemplate;

public class MQMessageSender {

	private JmsTemplate jmsTemplate;

	public void setJmsTemplate(JmsTemplate jmsTemplate) {
		this.jmsTemplate = jmsTemplate;
	}

	public void send(final MQMessage msg) {
		jmsTemplate.convertAndSend(msg);
	}

}

메시지 Converter

package com.iot.mq;

import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.Session;

import org.springframework.jms.support.converter.MessageConversionException;
import org.springframework.jms.support.converter.MessageConverter;

public class MQMessageConverter implements MessageConverter {

	public MQMessageConverter() {

	}

	public Object fromMessage(Message message) throws JMSException, MessageConversionException {
		if (!(message instanceof MapMessage)) {
			throw new MessageConversionException("not MapMessage");
		}

		System.out.println("MsgConverterImpl.fromMessage");

		MapMessage mapMessage = (MapMessage) message;
		MQMessage msg = new MQMessage();
		msg.setName(mapMessage.getString("name"));
		msg.setValue(mapMessage.getString("value"));

		return msg;
	}

	public Message toMessage(Object object, Session session) throws JMSException, MessageConversionException {
		if (!(object instanceof MQMessage)) {
			throw new MessageConversionException("not Msg");
		}

		System.out.println("MsgConverterImpl.toMessage");

		MQMessage msg = (MQMessage) object;
		MapMessage mapMessage = session.createMapMessage();

		mapMessage.setString("name", msg.getName());
		mapMessage.setString("value", msg.getValue());

		return mapMessage;
	}
}

메시지 Driven POJO

package com.iot.mq;

public class MQMessageDrivenPOJO {
	
	public void handleMessage(MQMessage msg) {
		// handles the message
		System.out.println("handler called");
		System.out.println(msg);
	}
}

MQMessageController

package com.iot.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.ui.ModelMap;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;

import com.iot.mq.MQMessage;
import com.iot.mq.MQMessageSender;

@Controller
public class MQMessageController {

	@Autowired
	private MQMessageSender mqSender;

	@ResponseBody
	@RequestMapping(value = "/mq.iot", method = RequestMethod.GET)
	public String login(
			@RequestParam("name") String name, 
			@RequestParam("value") String value, 
			ModelMap modelMap) throws Exception 
	{

		MQMessage msg = new MQMessage();
		msg.setName(name);
		msg.setValue(value);

		// 전송
		mqSender.send(msg);
		System.out.println(msg + " sent");

		return "MQ Send success";
	}

}

iot-mq.xml

<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:amq="http://activemq.apache.org/schema/core" 
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="
	http://www.springframework.org/schema/beans 
	http://www.springframework.org/schema/beans/spring-beans-2.0.xsd 
	http://activemq.apache.org/schema/core 
	http://activemq.apache.org/schema/core/activemq-core.xsd">

	<!-- a pooling based JMS provider -->
	<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
		<property name="brokerURL">
			<value>vm://localhost</value>
		</property>
	</bean>

	<bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">
		<property name="connectionFactory" ref="connectionFactory" />
	</bean>

	<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
		<constructor-arg value="com.iot.mq" />
	</bean>

	<bean id="msgConverter" class="com.iot.mq.MQMessageConverter" />
	<bean id="msgMdp" class="com.iot.mq.MQMessageDrivenPOJO" />

	<bean id="purePojoMdp" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
		<property name="delegate" ref="msgMdp" />
		<property name="defaultListenerMethod" value="handleMessage" />
		<property name="messageConverter" ref="msgConverter" />
	</bean>

	<bean class="org.springframework.jms.listener.SimpleMessageListenerContainer">
		<property name="connectionFactory" ref="jmsFactory" />
		<property name="destination" ref="queueDestination" />
		<property name="messageListener" ref="purePojoMdp" />
		<property name="concurrentConsumers" value="3" /> <!-- Queue 개수 -->
	</bean>

	<!-- Spring JMS Template -->
	<bean id="producerJmsTemplate" class="org.springframework.jms.core.JmsTemplate">
		<property name="connectionFactory" ref="jmsFactory" />
		<property name="defaultDestination" ref="queueDestination" />
		<property name="messageConverter" ref="msgConverter" />
	</bean>

	<bean id="mqSender" class="com.iot.mq.MQMessageSender">
		<property name="jmsTemplate" ref="producerJmsTemplate" />
	</bean>
</beans>

실행

실행결과

MsgConverterImpl.toMessage
MQMessage [name=a, value=a1] sent
MsgConverterImpl.fromMessage
handler called
MQMessage [name=a, value=a1]