ActiveMQ一 入门

启动：\apache-activemq-5.6.0\bin\win32\activemq.bat

工程中引入 \apache-activemq-5.6.0\lib下的jar

Listener 消息消费者：

public class TopicListener implements MessageListener {

..............

       main()方法{

        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(url);

        connection = factory.createConnection();

        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        topic = session.createTopic("topictest.messages");   //创建一个topic

 

        MessageConsumer consumer = session.createConsumer(topic);  //创建topic的消费者

        consumer.setMessageListener(this);

 

        connection.start();

 

        producer = session.createProducer(control);

        System.out.println("Waiting for messages...");

}

    //消息处理方法

    @Override

    public void onMessage(Message message) {

        if (checkText(message, "SHUTDOWN")) {

 

            try {

                connection.close();

            } catch (Exception e) {

                e.printStackTrace(System.out);

            }

 

        } else if (checkText(message, "REPORT")) {

            // send a report:

            try {

                long time = System.currentTimeMillis() - start;

                String msg = "Received " + count + " in " + time + "ms";

                producer.send(session.createTextMessage(msg));

            } catch (Exception e) {

                e.printStackTrace(System.out);

            }

            count = 0;

 

        } else {

 

            if (count == 0) {

                start = System.currentTimeMillis();

            }

 

            if (++count % 1000 == 0) {

                System.out.println("Received " + count + " messages.");

            }

        }

    }

...................

}

 

消息生存者：

public class TopicPublisher{

    main(){

          ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(url);

        connection = factory.createConnection();

        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        topic = session.createTopic("topictest.messages");

 

        publisher = session.createProducer(topic);

        publisher.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

 

        payload = new byte[size];

        for (int i = 0; i < size; i++) {

            payload[i] = (byte)DATA[i % DATA.length];

        }

 

        session.createConsumer(control).setMessageListener(this);

        connection.start();

 

        long[] times = new long[batch];

        for (int i = 0; i < batch; i++) {

            if (i > 0) {

                Thread.sleep(delay * 1000);

            }

            times[i] = batch(messages);

            System.out.println("Batch " + (i + 1) + " of " + batch + " completed in " + times[i] + " ms.");

        }

 

        long min = min(times);

        long max = max(times);

        System.out.println("min: " + min + ", max: " + max + " avg: " + avg(times, min, max));

 

        // request shutdown

        publisher.send(session.createTextMessage("SHUTDOWN"));

 

        connection.stop();

        connection.close();

    }

    private long batch(int msgCount) throws Exception {

        long start = System.currentTimeMillis();

        remaining = subscribers;

        publish();

        waitForCompletion();

        return System.currentTimeMillis() - start;

    }

 

    private void publish() throws Exception {

 

        // send events

        BytesMessage msg = session.createBytesMessage();

        msg.writeBytes(payload);

        for (int i = 0; i < messages; i++) {

            publisher.send(msg);

            if ((i + 1) % 1000 == 0) {

                System.out.println("Sent " + (i + 1) + " messages");

            }

        }

 

        // request report

        publisher.send(session.createTextMessage("REPORT"));

    }

}

 管理页面地址： http://127.0.0.1:8161/admin/topics.jsp

测试结果监控：

 ===================以下为spring中的用法：

配置：

 

beans中引入activemq的schema：

<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.xsd

 http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.5.0.xsd">

 

创建queue/topic:（服务器启动时）

<amq:queue name="ORDER_TOPIC" physicalName="ActiveMQ.ORDER.${jms_node}" />

 

配置服务器连接工厂，参数是activemq服务器地址：

jms_server=tcp://192.168.0.238:61616?wireFormat.maxInactivityDuration=0

 

<!-- ActiveMQ connectionFactory -->

<amq:connectionFactory id="jmsConnectionFactory" brokerURL="${jms_server}" />

 

创建 发送消息的bean：

<bean id="myJmsTemplate" class="org.springframework.jms.core.JmsTemplate">

<property name="connectionFactory">

<bean class="org.springframework.jms.connection.SingleConnectionFactory">

<property name="targetConnectionFactory" ref="jmsConnectionFactory" />

</bean>

</property>

</bean>

 

<bean id="orderMessageProducer" class="com.xxx.comm.jms.TopicMessageProducer">

<property name="template" ref="myJmsTemplate" />

<property name="destination" value="ActiveMQ.ORDER" />

</bean>

再看看TopicMessageProducer的实现（pojo而已）：

public class TopicMessageProducer {

 

private static final Log log = LogFactory.getLog(TopicMessageProducer.class);

 

    private JmsTemplate template;

 

private String destination;

 

    public void sendMsg(Message msg){

        log.info(msg);

    String[] queues = MessageProtector.getInstance().getQueues(destination);

    for (String string : queues) {

        template.convertAndSend(string, msg);

}

    }

    

public void setTemplate(JmsTemplate template) {

this.template = template;

}

 

public void setDestination(String destination) {

this.destination = destination;

}

}

其中Message 是自定义的一个pojo，但须实现 Serializable：

public class Message implements Serializable{

   private int a;

   ......

}

 

发送消息： 

getBean("orderMessageProducer").sendMsg(new Message());