Java编程rabbitMQ实现消息的收发

<!-- for rabbitmq --> 	<dependency>		<groupId>com.rabbitmq</groupId>		<artifactId>amqp-client</artifactId>		<version>2.8.2</version>	</dependency>	<dependency>		<groupId>org.springframework.amqp</groupId>		<artifactId>spring-amqp</artifactId>		<version>1.1.1.RELEASE</version>	</dependency>	<dependency>		<groupId>org.springframework.amqp</groupId>		<artifactId>spring-rabbit</artifactId>		<version>1.1.1.RELEASE</version>	</dependency>	<dependency>		<groupId>com.caucho</groupId>		<artifactId>hessian</artifactId>		<version>4.0.7</version>	</dependency> </dependencies>

public class EventMessage implements Serializable{	private String queueName;	private String exchangeName;	private byte[] eventData;	public EventMessage(String queueName, String exchangeName, byte[] eventData) {		this.queueName = queueName;		this.exchangeName = exchangeName;		this.eventData = eventData;	}	public EventMessage() {	}		public String getQueueName() {		return queueName;	}	public String getExchangeName() {		return exchangeName;	}	public byte[] getEventData() {		return eventData;	}	@Override	public String toString() {		return "EopEventMessage [queueName=" + queueName + ", exchangeName="				+ exchangeName + ", eventData=" + Arrays.toString(eventData)				+ "]";	}}

public interface CodecFactory {	byte[] serialize(Object obj) throws IOException;	Object deSerialize(byte[] in) throws IOException;}

public class HessionCodecFactory implements CodecFactory {	private final Logger logger = Logger.getLogger(HessionCodecFactory.class);	@Override	public byte[] serialize(Object obj) throws IOException {		ByteArrayOutputStream baos = null;		HessianOutput output = null;		try {			baos = new ByteArrayOutputStream(1024);			output = new HessianOutput(baos);			output.startCall();			output.writeObject(obj);			output.completeCall();		} catch (final IOException ex) {			throw ex;		} finally {			if (output != null) {				try {					baos.close();				} catch (final IOException ex) {					this.logger.error("Failed to close stream.", ex);				}			}		}		return baos != null ? baos.toByteArray() : null;	}	@Override	public Object deSerialize(byte[] in) throws IOException {		Object obj = null;		ByteArrayInputStream bais = null;		HessianInput input = null;		try {			bais = new ByteArrayInputStream(in);			input = new HessianInput(bais);			input.startReply();			obj = input.readObject();			input.completeReply();		} catch (final IOException ex) {			throw ex;		} catch (final Throwable e) {			this.logger.error("Failed to decode object.", e);		} finally {			if (input != null) {				try {					bais.close();				} catch (final IOException ex) {					this.logger.error("Failed to close stream.", ex);  }  } } return obj;	}}

public interface EventTemplate {	void send(String queueName,String exchangeName,Object eventContent) throws SendRefuseException;	void send(String queueName,String exchangeName,Object eventContent,CodecFactory codecFactory) throws SendRefuseException;}

public class DefaultEventTemplate implements EventTemplate {	private static final Logger logger = Logger.getLogger(DefaultEventTemplate.class);	private AmqpTemplate eventAmqpTemplate;	private CodecFactory defaultCodecFactory;//	private DefaultEventController eec;//	public DefaultEventTemplate(AmqpTemplate eopAmqpTemplate,//			CodecFactory defaultCodecFactory, DefaultEventController eec) {//		this.eventAmqpTemplate = eopAmqpTemplate;//		this.defaultCodecFactory = defaultCodecFactory;//		this.eec = eec;//	}	public DefaultEventTemplate(AmqpTemplate eopAmqpTemplate,CodecFactory defaultCodecFactory) {		this.eventAmqpTemplate = eopAmqpTemplate;		this.defaultCodecFactory = defaultCodecFactory;	}	@Override	public void send(String queueName, String exchangeName, Object eventContent)			throws SendRefuseException {		this.send(queueName, exchangeName, eventContent, defaultCodecFactory);	} 	@Override	public void send(String queueName, String exchangeName, Object eventContent,			CodecFactory codecFactory) throws SendRefuseException {		if (StringUtils.isEmpty(queueName) || StringUtils.isEmpty(exchangeName)) {			throw new SendRefuseException("queueName exchangeName can not be empty.");		}//		if (!eec.beBinded(exchangeName, queueName))//			eec.declareBinding(exchangeName, queueName);		byte[] eventContentBytes = null;		if (codecFactory == null) {			if (eventContent == null) {				logger.warn("Find eventContent is null,are you sure...");			} else {				throw new SendRefuseException(						"codecFactory must not be null ,unless eventContent is null");			}		} else {			try {				eventContentBytes = codecFactory.serialize(eventContent);			} catch (IOException e) {				throw new SendRefuseException(e);			}		}		// 构造成Message		EventMessage msg = new EventMessage(queueName, exchangeName,				eventContentBytes);		try {			eventAmqpTemplate.convertAndSend(exchangeName, queueName, msg);		} catch (AmqpException e) {			logger.error("send event fail. Event Message : [" + eventContent + "]", e);			throw new SendRefuseException("send event fail", e);		}	}}

public interface EventProcesser {	public void process(Object e);}

/** * MessageListenerAdapter的Pojo * <p>消息处理适配器，主要功能：</p> * <p>1、将不同的消息类型绑定到对应的处理器并本地缓存，如将queue01+exchange01的消息统一交由A处理器来出来</p> * <p>2、执行消息的消费分发，调用相应的处理器来消费属于它的消息</p> *  */public class MessageAdapterHandler {	private static final Logger logger = Logger.getLogger(MessageAdapterHandler.class);	private ConcurrentMap<String, EventProcessorWrap> epwMap;	public MessageAdapterHandler() {		this.epwMap = new ConcurrentHashMap<String, EventProcessorWrap>();	}	public void handleMessage(EventMessage eem) {		logger.debug("Receive an EventMessage: [" + eem + "]");		// 先要判断接收到的message是否是空的，在某些异常情况下，会产生空值		if (eem == null) {			logger.warn("Receive an null EventMessage, it may product some errors, and processing message is canceled.");			return;		}		if (StringUtils.isEmpty(eem.getQueueName()) || StringUtils.isEmpty(eem.getExchangeName())) {			logger.warn("The EventMessage's queueName and exchangeName is empty, this is not allowed, and processing message is canceled.");			return;		}		// 解码，并交给对应的EventHandle执行		EventProcessorWrap eepw = epwMap.get(eem.getQueueName()+"|"+eem.getExchangeName());		if (eepw == null) {			logger.warn("Receive an EopEventMessage, but no processor can do it.");			return;		}		try {			eepw.process(eem.getEventData());		} catch (IOException e) {			logger.error("Event content can not be Deserialized, check the provided CodecFactory.",e);			return;		}	}	protected void add(String queueName, String exchangeName, EventProcesser processor,CodecFactory codecFactory) {		if (StringUtils.isEmpty(queueName) || StringUtils.isEmpty(exchangeName) || processor == null || codecFactory == null) {			throw new RuntimeException("queueName and exchangeName can not be empty,and processor or codecFactory can not be null. ");		}		EventProcessorWrap epw = new EventProcessorWrap(codecFactory,processor);		EventProcessorWrap oldProcessorWrap = epwMap.putIfAbsent(queueName + "|" + exchangeName, epw);		if (oldProcessorWrap != null) {			logger.warn("The processor of this queue and exchange exists, and the new one can't be add");		}	}	protected Set<String> getAllBinding() {		Set<String> keySet = epwMap.keySet();		return keySet;	}	protected static class EventProcessorWrap {		private CodecFactory codecFactory;		private EventProcesser eep;		protected EventProcessorWrap(CodecFactory codecFactory,				EventProcesser eep) {			this.codecFactory = codecFactory;			this.eep = eep;		}		public void process(byte[] eventData) throws IOException{			Object obj = codecFactory.deSerialize(eventData);			eep.process(obj);		}	}}

public class MessageErrorHandler implements ErrorHandler{	private static final Logger logger = Logger.getLogger(MessageErrorHandler.class);	@Override	public void handleError(Throwable t) {		logger.error("RabbitMQ happen a error:" + t.getMessage(), t);	}}

public class EventControlConfig {	private final static int DEFAULT_PORT = 5672;	private final static String DEFAULT_USERNAME = "guest";	private final static String DEFAULT_PASSWORD = "guest";	private final static int DEFAULT_PROCESS_THREAD_NUM = Runtime.getRuntime().availableProcessors() * 2;	private static final int PREFETCH_SIZE = 1;	private String serverHost ;	private int port = DEFAULT_PORT;	private String username = DEFAULT_USERNAME;	private String password = DEFAULT_PASSWORD;	private String virtualHost;	/**	 * 和rabbitmq建立连接的超时时间	 */	private int connectionTimeout = 0;	/**	 * 事件消息处理线程数，默认是 CPU核数 * 2	 */	private int eventMsgProcessNum;	/**	 * 每次消费消息的预取值	 */	private int prefetchSize;		public EventControlConfig(String serverHost) {		this(serverHost,DEFAULT_PORT,DEFAULT_USERNAME,DEFAULT_PASSWORD,null,0,DEFAULT_PROCESS_THREAD_NUM,DEFAULT_PROCESS_THREAD_NUM,new HessionCodecFactory());	}	public EventControlConfig(String serverHost, int port, String username,			String password, String virtualHost, int connectionTimeout,			int eventMsgProcessNum,int prefetchSize,CodecFactory defaultCodecFactory) {		this.serverHost = serverHost;		this.port = port>0?port:DEFAULT_PORT;		this.username = username;		this.password = password;		this.virtualHost = virtualHost;		this.connectionTimeout = connectionTimeout>0?connectionTimeout:0;		this.eventMsgProcessNum = eventMsgProcessNum>0?eventMsgProcessNum:DEFAULT_PROCESS_THREAD_NUM;		this.prefetchSize = prefetchSize>0?prefetchSize:PREFETCH_SIZE;	}	public String getServerHost() {		return serverHost;	}	public int getPort() {		return port;	}	public String getUsername() {		return username;	}	public String getPassword() {		return password;	}	public String getVirtualHost() {		return virtualHost;	}	public int getConnectionTimeout() {		return connectionTimeout;	}	public int getEventMsgProcessNum() {		return eventMsgProcessNum;	}	public int getPrefetchSize() {		return prefetchSize;	}}

public interface EventController {		/**	 * 控制器启动方法	 */	void start();	/**	 * 获取发送模版	 */	EventTemplate getEopEventTemplate();	/**	 * 绑定消费程序到对应的exchange和queue	 */	EventController add(String queueName, String exchangeName, EventProcesser eventProcesser);	/*in map, the key is queue name, but value is exchange name*/	EventController add(Map<String,String> bindings, EventProcesser eventProcesser);}

/** * 和rabbitmq通信的控制器，主要负责： * <p>1、和rabbitmq建立连接</p> * <p>2、声明exChange和queue以及它们的绑定关系</p> * <p>3、启动消息监听容器，并将不同消息的处理者绑定到对应的exchange和queue上</p> * <p>4、持有消息发送模版以及所有exchange、queue和绑定关系的本地缓存</p> * @author yangyong * */public class DefaultEventController implements EventController {	private CachingConnectionFactory rabbitConnectionFactory;	private EventControlConfig config;	private RabbitAdmin rabbitAdmin;	private CodecFactory defaultCodecFactory = new HessionCodecFactory();	private SimpleMessageListenerContainer msgListenerContainer; // rabbitMQ msg listener container	private MessageAdapterHandler msgAdapterHandler = new MessageAdapterHandler();	private MessageConverter serializerMessageConverter = new SerializerMessageConverter(); // 直接指定	//queue cache, key is exchangeName	private Map<String, DirectExchange> exchanges = new HashMap<String,DirectExchange>();	//queue cache, key is queueName	private Map<String, Queue> queues = new HashMap<String, Queue>();	//bind relation of queue to exchange cache, value is exchangeName | queueName	private Set<String> binded = new HashSet<String>();	private EventTemplate eventTemplate; // 给App使用的Event发送客户端	private AtomicBoolean isStarted = new AtomicBoolean(false);	private static DefaultEventController defaultEventController;	public synchronized static DefaultEventController getInstance(EventControlConfig config){		if(defaultEventController==null){			defaultEventController = new DefaultEventController(config);		}		return defaultEventController;	}	private DefaultEventController(EventControlConfig config){		if (config == null) {			throw new IllegalArgumentException("Config can not be null.");		}		this.config = config;		initRabbitConnectionFactory();		// 初始化AmqpAdmin		rabbitAdmin = new RabbitAdmin(rabbitConnectionFactory);		// 初始化RabbitTemplate		RabbitTemplate rabbitTemplate = new RabbitTemplate(rabbitConnectionFactory);		rabbitTemplate.setMessageConverter(serializerMessageConverter);		eventTemplate = new DefaultEventTemplate(rabbitTemplate,defaultCodecFactory, this);	}	/**	 * 初始化rabbitmq连接	 */	private void initRabbitConnectionFactory() {		rabbitConnectionFactory = new CachingConnectionFactory();		rabbitConnectionFactory.setHost(config.getServerHost());		rabbitConnectionFactory.setChannelCacheSize(config.getEventMsgProcessNum());		rabbitConnectionFactory.setPort(config.getPort());		rabbitConnectionFactory.setUsername(config.getUsername());		rabbitConnectionFactory.setPassword(config.getPassword());		if (!StringUtils.isEmpty(config.getVirtualHost())) {			rabbitConnectionFactory.setVirtualHost(config.getVirtualHost());		}	}	/**	 * 注销程序	 */	public synchronized void destroy() throws Exception {		if (!isStarted.get()) {			return;		}		msgListenerContainer.stop();		eventTemplate = null;		rabbitAdmin = null;		rabbitConnectionFactory.destroy();	}	@Override	public void start() {		if (isStarted.get()) {			return;		}		Set<String> mapping = msgAdapterHandler.getAllBinding();		for (String relation : mapping) {			String[] relaArr = relation.split("//|");			declareBinding(relaArr[1], relaArr[0]);		}		initMsgListenerAdapter();		isStarted.set(true);	}	/**	 * 初始化消息监听器容器	 */	private void initMsgListenerAdapter(){		MessageListener listener = new MessageListenerAdapter(msgAdapterHandler,serializerMessageConverter);		msgListenerContainer = new SimpleMessageListenerContainer();		msgListenerContainer.setConnectionFactory(rabbitConnectionFactory);		msgListenerContainer.setAcknowledgeMode(AcknowledgeMode.AUTO);		msgListenerContainer.setMessageListener(listener);		msgListenerContainer.setErrorHandler(new MessageErrorHandler());		msgListenerContainer.setPrefetchCount(config.getPrefetchSize()); // 设置每个消费者消息的预取值		msgListenerContainer.setConcurrentConsumers(config.getEventMsgProcessNum());		msgListenerContainer.setTxSize(config.getPrefetchSize());//设置有事务时处理的消息数		msgListenerContainer.setQueues(queues.values().toArray(new Queue[queues.size()]));		msgListenerContainer.start();	}	@Override	public EventTemplate getEopEventTemplate() {		return eventTemplate;	}	@Override	public EventController add(String queueName, String exchangeName,EventProcesser eventProcesser) {		return add(queueName, exchangeName, eventProcesser, defaultCodecFactory);	}		public EventController add(String queueName, String exchangeName,EventProcesser eventProcesser,CodecFactory codecFactory) {		msgAdapterHandler.add(queueName, exchangeName, eventProcesser, defaultCodecFactory);		if(isStarted.get()){			initMsgListenerAdapter();		}		return this;	}	@Override	public EventController add(Map<String, String> bindings,			EventProcesser eventProcesser) {		return add(bindings, eventProcesser,defaultCodecFactory);	}	public EventController add(Map<String, String> bindings,			EventProcesser eventProcesser, CodecFactory codecFactory) {		for(Map.Entry<String, String> item: bindings.entrySet()) 			msgAdapterHandler.add(item.getKey(),item.getValue(), eventProcesser,codecFactory);		return this;	}	/**	 * exchange和queue是否已经绑定	 */	protected boolean beBinded(String exchangeName, String queueName) {		return binded.contains(exchangeName+"|"+queueName);	}	/**	 * 声明exchange和queue已经它们的绑定关系	 */	protected synchronized void declareBinding(String exchangeName, String queueName) {		String bindRelation = exchangeName+"|"+queueName;		if (binded.contains(bindRelation)) return;				boolean needBinding = false;		DirectExchange directExchange = exchanges.get(exchangeName);		if(directExchange == null) {			directExchange = new DirectExchange(exchangeName, true, false, null);			exchanges.put(exchangeName, directExchange);			rabbitAdmin.declareExchange(directExchange);//声明exchange			needBinding = true;		}		Queue queue = queues.get(queueName);		if(queue == null) {			queue = new Queue(queueName, true, false, false);			queues.put(queueName, queue);			rabbitAdmin.declareQueue(queue);	//声明queue			needBinding = true;		}		if(needBinding) {			Binding binding = BindingBuilder.bind(queue).to(directExchange).with(queueName);//将queue绑定到exchange			rabbitAdmin.declareBinding(binding);//声明绑定关系			binded.add(bindRelation);		}	}}

@SuppressWarnings("serial")public class People implements Serializable{	private int id;	private String name;	private boolean male;	private People spouse;	private List<People> friends;	public int getId() {		return id;	}	public void setId(int id) {		this.id = id;	}	public String getName() {		return name;	}	public void setName(String name) {		this.name = name;	}	public boolean isMale() {		return male;	}	public void setMale(boolean male) {		this.male = male;	}	public People getSpouse() {		return spouse;	}	public void setSpouse(People spouse) {		this.spouse = spouse;	}	public List<People> getFriends() {		return friends;	}	public void setFriends(List<People> friends) {		this.friends = friends;	}	@Override	public String toString() {		// TODO Auto-generated method stub		return "People[id="+id+",name="+name+",male="+male+"]";	}}

public class RabbitMqTest{	private String defaultHost = "127.0.0.1";	private String defaultExchange = "EXCHANGE_DIRECT_TEST";	private String defaultQueue = "QUEUE_TEST";	private DefaultEventController controller;	private EventTemplate eventTemplate;	@Before	public void init() throws IOException{		EventControlConfig config = new EventControlConfig(defaultHost);		controller = DefaultEventController.getInstance(config);		eventTemplate = controller.getEopEventTemplate();		controller.add(defaultQueue, defaultExchange, new ApiProcessEventProcessor());		controller.start();	}	@Test	public void sendString() throws SendRefuseException{		eventTemplate.send(defaultQueue, defaultExchange, "hello world");	}		@Test	public void sendObject() throws SendRefuseException{		eventTemplate.send(defaultQueue, defaultExchange, mockObj());	}		@Test	public void sendTemp() throws SendRefuseException, InterruptedException{		String tempExchange = "EXCHANGE_DIRECT_TEST_TEMP";//以前未声明的exchange		String tempQueue = "QUEUE_TEST_TEMP";//以前未声明的queue		eventTemplate.send(tempQueue, tempExchange, mockObj());		//发送成功后此时不会接受到消息，还需要绑定对应的消费程序		controller.add(tempQueue, tempExchange, new ApiProcessEventProcessor());	}		@After	public void end() throws InterruptedException{		Thread.sleep(2000);	}		private People mockObj(){		People jack = new People();		jack.setId(1);		jack.setName("JACK");		jack.setMale(true);				List<People> friends = new ArrayList<>();		friends.add(jack);		People hanMeiMei = new People();		hanMeiMei.setId(1);		hanMeiMei.setName("韩梅梅");		hanMeiMei.setMale(false);		hanMeiMei.setFriends(friends);				People liLei = new People();		liLei.setId(2);		liLei.setName("李雷");		liLei.setMale(true);		liLei.setFriends(friends);		liLei.setSpouse(hanMeiMei);		hanMeiMei.setSpouse(liLei);		return hanMeiMei;	}	class ApiProcessEventProcessor implements EventProcesser{		@Override		public void process(Object e) {//消费程序这里只是打印信息			Assert.assertNotNull(e);			System.out.println(e);			if(e instanceof People){				People people = (People)e;				System.out.println(people.getSpouse());				System.out.println(people.getFriends());			}		}	}}