package org.apache.qpid.contrib.json;

import org.apache.commons.configuration2.Configuration;
import org.apache.commons.configuration2.FileBasedConfiguration;
import org.apache.commons.configuration2.PropertiesConfiguration;
import org.apache.commons.configuration2.builder.FileBasedConfigurationBuilder;
import org.apache.commons.configuration2.builder.fluent.Parameters;
import org.apache.commons.configuration2.ex.ConfigurationException;
import org.apache.qpid.contrib.json.utils.BZip2Utils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;

/**
 * 
 * @author 张大川
 * ampq 4.0 改造
 */
public class SendMessageUtils {

	/**
	 * 
	 * 
	 * @param queueName
	 *            队列名称
	 * @param object
	 *            要传送的对象
	 * @throws Exception
	 */
	public static void sendMessage(String queueName, Object... object) throws Exception {
		Configuration config = null;
		boolean isCompress = false;
		Parameters params = new Parameters();
		FileBasedConfigurationBuilder<FileBasedConfiguration> builder = new FileBasedConfigurationBuilder<FileBasedConfiguration>(
				PropertiesConfiguration.class).configure(params.properties().setFileName("rabbitmq.properties"));
		try {
			config = builder.getConfiguration();

		} catch (ConfigurationException cex) {
			cex.printStackTrace();
		}
		ConnectionFactory factory = new ConnectionFactory();
		// System.out.println();
		factory.setHost(config.getString("hostname"));
		factory.setUsername(config.getString("username"));
		factory.setPassword(config.getString("password"));
		factory.useNio();
		if (config.containsKey("isCompress")) {
			isCompress = config.getBoolean("isCompress");
		}
		if (config.containsKey("port")) {
			factory.setPort(config.getInt("port"));
		} else {
			factory.setPort(AMQP.PROTOCOL.PORT);
		}

		Connection connection = factory.newConnection();
		Channel channel = connection.createChannel();

		channel.queueDeclare(queueName, true, false, false, null);

		String jsonMessage;
		Object[] messages = object;
		for (Object object2 : messages) {
			jsonMessage = JSON.toJSONString(object2, SerializerFeature.WriteClassName);

			if (isCompress) {
				channel.basicPublish("", queueName, MessageProperties.PERSISTENT_TEXT_PLAIN,
						BZip2Utils.compress(jsonMessage.getBytes()));
			} // 持久化消息

			else {
				channel.basicPublish("", queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, jsonMessage.getBytes());
			}
		}

		channel.close();
		connection.close();

	}

}

@@ -438,62 +438,8 @@
 ion;
-%0Aimport org.apache.qpid.contrib.json.utils.BZip2Utils;
 %0A%0Aim
@@ -758,20 +758,8 @@
 %E5%B7%9D%0A *
- ampq 4.0 %E6%94%B9%E9%80%A0
 %0A */
@@ -1037,38 +1037,9 @@
 ll;%0A
-%09%09boolean isCompress = false;
 %0A
+
 %09%09Pa
@@ -1656,24 +1656,25 @@
 sword%22));%0A%09%09
+ 
 factory.useN
@@ -1683,102 +1683,8 @@
 ();%0A
-%09%09if (config.containsKey(%22isCompress%22)) %7B%0A%09%09%09isCompress = config.getBoolean(%22isCompress%22);%0A%09%09%7D
 %0A%09%09i
@@ -2149,188 +2149,8 @@
 );%0A%0A
-%09%09%09if (isCompress) %7B%0A%09%09%09%09channel.basicPublish(%22%22, queueName, MessageProperties.PERSISTENT_TEXT_PLAIN,%0A%09%09%09%09%09%09BZip2Utils.compress(jsonMessage.getBytes()));%0A%09%09%09%7D // %E6%8C%81%E4%B9%85%E5%8C%96%E6%B6%88%E6%81%AF%0A%0A%09%09%09else %7B%0A%09
 %09%09%09c
@@ -2254,12 +2254,8 @@
 ));%0A
-%09%09%09%7D
 %0A%09%09%7D

Update SendMessageUtils.java