项目中用到了ActiveMQ，本来是用spring方式写的，现在转换成jFinal给大家参考

只是一个粗略的插件，没有更严谨，大家可以自己改改

这个插件可以实现多链接(内部也有连接池)、多消息类型的发送

首先我是maven项目，加入依赖

<dependency>
    <groupId>com.jfinal</groupId>
    <artifactId>jfinal</artifactId>
    <version>2.2</version>
</dependency>
<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-pool</artifactId>
    <version>5.14.0</version>
</dependency>

定义Sender类

package jfinal.plugin.activemq;
import org.apache.activemq.pool.PooledConnection;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.Topic;
/**
 * 消息发送者
 *
 * @author 孙宇
 */
public class JmsSender {
    private String name;
    private Session session;
    private MessageProducer producer;
    public JmsSender(String name,
                     PooledConnection connection,
                     Destination type,
                     String subject) throws JMSException {
        this.name = name;
        // 事务性会话，自动确认消息
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 消息的目的地（Queue/Topic）
        if (type.equals(Destination.Topic)) {
            Topic destination = session.createTopic(subject);
            producer = session.createProducer(destination);
        } else {
            Queue destination = session.createQueue(subject);
            producer = session.createProducer(destination);
        }
        // 不持久化消息
        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
    }
    public String getName() {
        return name;
    }
    public Session getSession() {
        return session;
    }
    public void sendMessage(Message message) throws JMSException {
        producer.send(message);
    }
}

定义Receiver类

package jfinal.plugin.activemq;
import org.apache.activemq.pool.PooledConnection;
import javax.jms.BytesMessage;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.StreamMessage;
import javax.jms.TextMessage;
import javax.jms.Topic;
import java.util.Enumeration;
/**
 * 消息接收者
 *
 * @author 孙宇
 */
public class JmsReceiver implements MessageListener {
    private String name;
    private Session session;
    private MessageConsumer consumer;
    public JmsReceiver(String name,
                       PooledConnection connection,
                       Destination type,
                       String subject) throws JMSException {
        this.name = name;
        // 事务性会话，自动确认消息
        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 消息的目的地（Queue/Topic）
        if (type.equals(Destination.Topic)) {
            Topic destination = session.createTopic(subject);
            consumer = session.createConsumer(destination);
        } else {
            Queue destination = session.createQueue(subject);
            consumer = session.createConsumer(destination);
        }
        consumer.setMessageListener(this);
    }
    public String getName() {
        return name;
    }
    @Override
    public void onMessage(Message message) {
        try {
            if (message instanceof TextMessage) {
                TextMessage msg = (TextMessage) message;
                System.out.println(msg.getText());
            } else if (message instanceof MapMessage) {
                MapMessage msg = (MapMessage) message;
                Enumeration enumer = msg.getMapNames();
                while (enumer.hasMoreElements()) {
                    Object obj = enumer.nextElement();
                    System.out.println(msg.getObject(obj.toString()));
                }
            } else if (message instanceof StreamMessage) {
                StreamMessage msg = (StreamMessage) message;
                System.out.println(msg.readString());
                System.out.println(msg.readBoolean());
                System.out.println(msg.readLong());
            } else if (message instanceof ObjectMessage) {
                ObjectMessage msg = (ObjectMessage) message;
                System.out.println(msg);
            } else if (message instanceof BytesMessage) {
                BytesMessage msg = (BytesMessage) message;
                byte[] byteContent = new byte[1024];
                int length = -1;
                StringBuffer content = new StringBuffer();
                while ((length = msg.readBytes(byteContent)) != -1) {
                    content.append(new String(byteContent, 0, length));
                }
                System.out.println(content.toString());
            } else {
                System.out.println(message);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

定义类型

package jfinal.plugin.activemq;
/**
 * 目标类型
 *
 * @author 孙宇
 */
public enum Destination {
    Queue, Topic
}

activeMq插件

package jfinal.plugin.activemq;
import com.jfinal.plugin.IPlugin;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.pool.PooledConnection;
import org.apache.activemq.pool.PooledConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.jms.JMSException;
/**
 * @author 孙宇
 */
public class ActiveMQPlugin implements IPlugin {
    private static final Logger logger = LoggerFactory.getLogger(Thread.currentThread().getClass());
    private String url;
    private String name;
    public ActiveMQPlugin(String url,
                          String name) {
        this.url = url;
        this.name = name;
    }
    public ActiveMQPlugin(String url) {
        this.url = url;
        this.name = ActiveMQ.defaultName;
    }
    @Override
    public boolean start() {
        logger.info("初始化activeMQ配置");
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory();
        activeMQConnectionFactory.setUserName(ActiveMQConnection.DEFAULT_USER);
        activeMQConnectionFactory.setPassword(ActiveMQConnection.DEFAULT_PASSWORD);
        activeMQConnectionFactory.setBrokerURL(url);
        activeMQConnectionFactory.setDispatchAsync(true);//异步发送消息
        PooledConnectionFactory pooledConnectionFactory = new PooledConnectionFactory(activeMQConnectionFactory);
        pooledConnectionFactory.setMaximumActiveSessionPerConnection(200);
        pooledConnectionFactory.setIdleTimeout(120);
        pooledConnectionFactory.setMaxConnections(5);
        pooledConnectionFactory.setBlockIfSessionPoolIsFull(true);
        try {
            PooledConnection connection = (PooledConnection) pooledConnectionFactory.createConnection();
            connection.start();
            ActiveMQ.pooledConnectionMap.put(name, connection);
        } catch (JMSException e) {
            e.printStackTrace();
        }
        return true;
    }
    @Override
    public boolean stop() {
        return true;
    }
}

activeMQ工具

package jfinal.plugin.activemq;
import org.apache.activemq.pool.PooledConnection;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.ConcurrentHashMap;
/**
 * @author 孙宇
 */
public class ActiveMQ {
    public static final ConcurrentHashMap<String, PooledConnection> pooledConnectionMap = new ConcurrentHashMap<>();
    public static final ConcurrentHashMap<String, JmsSender> senderMap = new ConcurrentHashMap<>();
    public static final ConcurrentHashMap<String, JmsReceiver> receiverMap = new ConcurrentHashMap<>();
    public static final String defaultName = "main";
    private static final Logger logger = LoggerFactory.getLogger(Thread.currentThread().getClass());
    public static void addSender(JmsSender sender) {
        senderMap.put(sender.getName(), sender);
    }
    public static JmsSender getSender(String name) {
        return senderMap.get(name);
    }
    public static void addReceiver(JmsReceiver receiver) {
        receiverMap.put(receiver.getName(), receiver);
    }
    public static JmsReceiver getReceiver(String name) {
        return receiverMap.get(name);
    }
    public static void addConnection(String connectionName,
                                     PooledConnection connection) {
        pooledConnectionMap.put(connectionName, connection);
    }
    public static PooledConnection getConnection() {
        return pooledConnectionMap.get(defaultName);
    }
    public static PooledConnection getConnection(String connectionName) {
        return pooledConnectionMap.get(connectionName);
    }
}

使用的时候，

ActiveMQPlugin p = new ActiveMQPlugin("failover://(tcp://127.0.0.1:61616)?initialReconnectDelay=1000");
p.start();

当然，这样只是创建了链接池，但是没有生产者和消费者
还要添加这两个对象

String subject = "test";
ActiveMQ.addSender(new JmsSender("testSender1", ActiveMQ.getConnection(), Destination.Queue, subject));//定义发送者
ActiveMQ.addReceiver(new JmsReceiver("testReceiver1", ActiveMQ.getConnection(), Destination.Queue, subject));//定义接受者

发送一个消息

JmsSender sq1 = ActiveMQ.getSender("testSender1");
TextMessage msg = sq1.getSession().createTextMessage("测试" + new Date());
sq1.sendMessage(msg);

我里面定义了很多消息类型，不一定非要文本型
一个测试类

package test.jfinal.plugin.activemq;
import jfinal.plugin.activemq.ActiveMQ;
import jfinal.plugin.activemq.ActiveMQPlugin;
import jfinal.plugin.activemq.Destination;
import jfinal.plugin.activemq.JmsReceiver;
import jfinal.plugin.activemq.JmsSender;
import javax.jms.JMSException;
import javax.jms.TextMessage;
import java.util.Date;
/**
 * @author 孙宇
 */
public class TestActiveMQPlugin {
    public static void main(String[] args) throws JMSException {
        ActiveMQPlugin p = new ActiveMQPlugin("failover://(tcp://127.0.0.1:61616)?initialReconnectDelay=1000");
        p.start();
        String subject = "test";
        ActiveMQ.addSender(new JmsSender("testSender1", ActiveMQ.getConnection(), Destination.Queue, subject));//定义发送者
        ActiveMQ.addReceiver(new JmsReceiver("testReceiver1", ActiveMQ.getConnection(), Destination.Queue, subject));//定义接受者
        for (int i = 0; i < 10; i++) {
            new Runnable() {
                @Override
                public void run() {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    try {
                        JmsSender sq1 = ActiveMQ.getSender("testSender1");
                        TextMessage msg = sq1.getSession().createTextMessage("测试" + new Date());
                        sq1.sendMessage(msg);
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            }.run();
        }
    }
}

有兴趣你也可以下载源码跑一下,执行TestActiveMQPlugin类就行

https://git.oschina.net/sypro/jfinalplugin.git

想看spring版的，看这个源码

https://git.oschina.net/sypro/demo.git