activeMQ简单实现

一、pom.xml

  <!--ActiveMQ-->
        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-all</artifactId>
            <version>5.8.0</version>
        </dependency>

二、下载服务器端

http://activemq.apache.org/activemq-580-release.html

 

三、运行服务器端

 

四、自己在代码中写生产者

package com.kf.ActiveMQ.queue;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * 消息生产者
 *
 */
class AppProducer {

    private static final String url = "tcp://LeiDaGou:61616"; // 消息发送的服务地址。61616为服务默认端口
    private static final String queueName = "queue-test"; // 注册的队列名字

    public static void main(String[] args) {
        try {
            // 1.创建ConnectionFactory
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);

            // 2.创建Connection
            Connection connection = null;
            connection = connectionFactory.createConnection();

            // 3.启动链接
            connection.start();

            // 4.创建回话
            // false:是否使用事务，AUTO_ACKNOWLEDGE:设置为自动应答的方式
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            // 5.创建一个目标
            Destination destination = session.createQueue(queueName);

            //6.创建一个生产者
            MessageProducer producer = session.createProducer(destination);

            for (int i = 0; i < 100; i++) {
                // 7.创建消息
                TextMessage textMessage = session.createTextMessage("test" + i);

                // 8.发布消息
                producer.send(textMessage);
                System.out.println("发送消息" + textMessage.getText());

            }
            // 9.关闭链接
            connection.close();

        } catch (JMSException e) {
            e.printStackTrace();
        }


    }


}

 五、消费者

package com.kf.ActiveMQ.queue;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * 消费者
 *
 */
public class AppConsumer {
    private static final String url = "tcp://LeiDaGou:61616"; // 消息发送的服务地址。61616为服务默认端口
    private static final String queueName = "queue-test"; // 注册的队列名字

    public static void main(String[] args) {
        try {
            // 1.创建ConnectionFactory
            ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);

            // 2.创建Connection
            Connection connection = null;
            connection = connectionFactory.createConnection();

            // 3.启动链接
            connection.start();

            // 4.创建回话
            // false:是否使用事务，AUTO_ACKNOWLEDGE:设置为自动应答的方式
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            // 5.创建一个目标
            Destination destination = session.createQueue(queueName);

            //6.创建一个消费者
            MessageConsumer consumer = session.createConsumer(destination);

            //7.创建一个监听器
            consumer.setMessageListener(new MessageListener() {
                public void onMessage(Message message) {
                    TextMessage textMessage = (TextMessage) message;
                    try {
                        System.out.println("接收消息" + textMessage.getText());
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
            /*// 8.关闭链接
            connection.close();*/
        } catch (JMSException e) {
            e.printStackTrace();
        }


    }
}

上面是queue类型的、topic类型的修改第五步

六、测试结果