ActiveMQ安装和使用

ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
factory.setTrustedPackages(new ArrayList(Arrays.asList("org.apache.activemq.test,org.apache.camel.test".split(","))));

<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-all</artifactId>
    <version>5.15.9</version>
</dependency>

public class ActiveMQ_Producer {
        public void sendMessage2ActiveMQ(String msg) {
        //连接工厂对象
        ConnectionFactory connectionFactory=null;
        //链接对象
        Connection connection=null;
        //会话对象
        Session session=null;
        //发送地址（即队列）
        Destination destination=null;
        //消息生产者
        MessageProducer messageProducer=null;
        //消息对象
        Message message=null;
        try {
             * 用户名和密码不同于服务器登录账号密码，该账号密码在conf/jetty-realm.properties中设置，用于链接ActiveMQ，上传消息的
             * 默认为admin：admin
             * userName:
             * password:
             * 在conf/activmq.xml中配置
             * brokerURL:协议://ip地址:端口
            connectionFactory=new ActiveMQConnectionFactory("amdin","admin", "tcp://192.168.40.128:61616");
            //创建链接
            connection=connectionFactory.createConnection();
            //启动链接
            connection.start();
            //创建session
             * transacted:是否开启事务  true|false
             *          true：开启事务，但后面的acknowledgeMode必须为Session.SESSION_TRANSACTED
             *          false：不开启事务
             * 区别：开启事务后，会保持原子性,通过commit和rollback方法，实现提交和回滚，
             * acknowledgeMode:    服务器接收到消息并处理后返回ack，客户端接收服务器到服务器发送的ack后知道已经处理完毕，不需要再从夫发送
             *      Session.AUTO_ACKNOWLEDGE    消息自动签收
             *      Session.CLIENT_ACKNOWLEDGE  客户端调用acknowledge方法手动签收
             *      Session.DUPS_OK_ACKNOWLEDGE 不必必须签收，消息可能会重复发送。在第二次重新传递消息的时候，消息头的
             *                                  JmsDelivered会被置为true标示当前消息已经传送过一次，客户端需要进行消息
             *                                  的重复处理控制。
             session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
             //创建地址，即队列的名称
             destination=session.createQueue("MassageTest");
             //创建生产者
             messageProducer = session.createProducer(destination);
             //创建消息(简单文本消息)
             message = session.createTextMessage(msg);
             //发送消息
             messageProducer.send(message);
        } catch (JMSException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }finally {
            //关闭资源
            if(messageProducer!=null) {
                try {
                    messageProducer.close();
                } catch (JMSException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
            if(session!=null) {
                try {
                    session.close();
                } catch (JMSException e) {
                    // TODO Auto-generated catch block





    
                    e.printStackTrace();
            if(connection!=null) {
                try {
                    connection.close();
                } catch (JMSException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
}

public class ActiveMQ_Consumer {
    public void getMessageFromActiveMQ() {
        //连接工厂对象
        ConnectionFactory connectionFactory=null;
        //链接对象
        Connection connection=null;
        //会话对象
        Session session=null;
        //发送地址（即队列）
        Destination destination=null;
        //消息消费者
        MessageConsumer messageConsumer=null;
        //消息对象
        TextMessage message=null;
        try {
            connectionFactory=new ActiveMQConnectionFactory("amdin","admin", "tcp://192.168.40.128:61616");
            //创建链接
            connection=connectionFactory.createConnection();
            //启动链接
            connection.start();
            //创建session
             session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
             //创建地址，即队列的名称
             destination=session.createQueue("MassageTest");
             //创建消费者
             messageConsumer = session.createConsumer(destination);
             //接受消息，并转换为TextMessage文本消息
             message = (TextMessage)messageConsumer.receive();
            //输出消息
             System.out.println("接收到："+message.getText());
        } catch (JMSException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }finally {
            //关闭资源
            if(messageConsumer!=null) {
                try {
                    messageConsumer.close();
                } catch (JMSException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
            if(session!=null) {
                try {
                    session.close();
                } catch (JMSException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
            if(connection!=null) {
                try {
                    connection.close();
                } catch (JMSException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
}

public static void main(String[] args) {
        ActiveMQ_Producer producer=new ActiveMQ_Producer();
        producer.sendMessage2ActiveMQ("helloword");
    }

public static void main(String[] args) {
        ActiveMQ_Consumer consumer=new ActiveMQ_Consumer();
        consumer.getMessageFromActiveMQ();
    }

public class ActiveMQ_Producer2 {
    public void sendMessage2ActiveMQ(Serializable obj) {
        //连接工厂对象
        ConnectionFactory connectionFactory=null;
        //链接对象
        Connection connection=null;
        //会话对象
        Session session=null;
        //发送地址（即队列）
        Destination destination=null;
        //消息生产者
        MessageProducer messageProducer=null;
        //消息对象
        Message message=null;
        try {
             * 用户名和密码不同于服务器登录账号密码，该账号密码在conf/jetty-realm.properties中设置，用于链接ActiveMQ，上传消息的
             * 默认为admin：admin
             * userName:
             * password:
             * 在conf/activmq.xml中配置
             * brokerURL:协议://ip地址:端口
            connectionFactory=new ActiveMQConnectionFactory("amdin","admin", "tcp://192.168.40.128:61616");
            //创建链接
            connection=connectionFactory.createConnection();
            //启动链接
            connection.start();
            //创建session
             * transacted:是否开启事务  true|false
             *          true：开启事务，但后面的acknowledgeMode必须为Session.SESSION_TRANSACTED
             *          false：不开启事务
             * 区别：开启事务后，会保持原子性,通过commit和rollback方法，实现提交和回滚，
             * acknowledgeMode:    服务器接收到消息并处理后返回ack，客户端接收服务器到服务器发送的ack后知道已经处理完毕，不需要再从夫发送
             *      Session.AUTO_ACKNOWLEDGE    消息自动签收
             *      Session.CLIENT_ACKNOWLEDGE  客户端调用acknowledge方法手动签收
             *      Session.DUPS_OK_ACKNOWLEDGE 不必必须签收，消息可能会重复发送。在第二次重新传递消息的时候，消息头的
             *                                  JmsDelivered会被置为true标示当前消息已经传送过一次，客户端需要进行消息
             *                                  的重复处理控制。
             session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
             //创建地址，即队列的名称
             destination=session.createQueue("ObjectMassageTest");
             //创建生产者（重点）
             messageProducer = session.createProducer(destination);
             //创建消息(简单文本消息)
             message = session.createObjectMessage(obj);
             //发送消息
             messageProducer.send(message);
        } catch (JMSException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }finally {




    

            //关闭资源
            if(messageProducer!=null) {
                try {
                    messageProducer.close();
                } catch (JMSException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
            if(session!=null) {
                try {
                    session.close();
                } catch (JMSException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
            if(connection!=null) {
                try {
                    connection.close();
                } catch (JMSException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
}

public class ActiveMQ_Consumer2 {
    public void getMessageFromActiveMQ() {
        //连接工厂对象
        ConnectionFactory connectionFactory=null;
        //链接对象
        Connection connection=null;
        //会话对象
        Session session=null;
        //发送地址（即队列）
        Destination destination=null;
        //消息消费者
        MessageConsumer messageConsumer=null;
        //消息对象
        Message message=null;
        try {
            connectionFactory=new ActiveMQConnectionFactory("amdin","admin", "tcp://192.168.40.128:61616");
            //在ActiveMQ5.12以后为了避免收到恶意代码，引入了安全机制，只允许指定的包里的对象能够被传输。
            //使用setTrustAllPackages方法设置所有的类都能够被传输
            //或者使指定的包里的类能够被传输
             * ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
                factory.setTrustedPackages(new ArrayList(Arrays.asList("org.apache.activemq.test,org.apache.camel.test".split(","))));
            ((ActiveMQConnectionFactory) connectionFactory).setTrustAllPackages(true);
            //创建链接
            connection=connectionFactory.createConnection();
            //启动链接
            connection.start();
            //创建session
             session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
             //创建地址，即队列的名称
             destination=session.createQueue("ObjectMassageTest");
             //创建消费者
             messageConsumer = session.createConsumer(destination);
             //接受消息
             message = messageConsumer.receive();
            //处理消息(重点)
             ObjectMessage objMessage=(ObjectMessage)message;
             User user = (User)objMessage.getObject();
             System.out.println(user.toString());
        } catch (JMSException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }finally {
            //关闭资源
            if(messageConsumer!=null) {
                try {
                    messageConsumer.close();
                } catch (JMSException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
            if(session!=null) {
                try {
                    session.close();
                } catch (JMSException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
            if(connection!=null) {
                try {
                    connection.close();
                } catch (JMSException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
}

//创建监听，处理消息
messageConsumer.setMessageListener(new MessageListener() {
    @Override
    public void onMessage(Message message) {
        //输出消息
        try {
            System.out.println("接收到："+((TextMessage)message).getText());
        } catch (JMSException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
});

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>cn.xyl</groupId>
    <artifactId>parent-ActiveMQ</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <packaging>pom</packaging>
    <!-- 对依赖的jar包的版本统一进行定义 -->
    <properties>
        <activemq.version>5.15.9</activemq.version>
        <xbean.version>4.5</xbean.version>
        <jms.version>4.3.22.RELEASE</jms.version>
        <activemq-pool.version>5.15.9</activemq-pool.version>
        <solrj.version>4.10.3</solrj.version>
        <jedis.version>2.9.0</jedis.version>
        <junit.version>4.12</junit.version>
        <spring.version>4.3.22.RELEASE</spring.version>
        <mybatis.version>3.2.8</mybatis.version>
        <mybatis.spring.version>1.2.2</mybatis.spring.version>
        <mysql.version>5.1.32</mysql.version>
        <slf4j.version>1.6.4</slf4j.version>
        <druid.version>1.0.9</druid.version>
        <jstl.version>1.2</jstl.version>
        <servlet-api.version>2.5</servlet-api.version>
        <tomcat.version>2.2</tomcat.version>
        <jsp-api.version>2.0</jsp-api.version>
        <zkClient-version>0.10</zkClient-version>
        <dubbo-version>2.5.4</dubbo-version>
        <jackson.version>2.4.2</jackson.version>
        <commons-net.version>3.3</commons-net.version>
        <commons-fileupload.version>1.3.1</commons-fileupload.version>
    </properties>
    <!-- jar包的依赖注入 ，由于该工程是一个父工程，所以jar包在该pom文件中只是声明 -->
    <dependencyManagement>
        <dependencies>
            <!-- ActiveMQ客户端完整jar包依赖 -->
            <dependency>
                <groupId>org.apache.activemq</groupId>
                <artifactId>activemq-all</artifactId>
                <version>${activemq.version}</version>
            </dependency>
            <!-- ActiveMQ和Spring整合配置文件标签处理jar包依赖 -->
            <dependency>
                <groupId>org.apache.xbean</groupId>
                <artifactId>xbean-spring</artifactId>
                <version>${xbean.version}</version>
            </dependency>
            <!-- Spring-JMS插件相关jar包依赖 -->




    

            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-jms</artifactId>
                <version>${jms.version}</version>
            </dependency>
            <!-- Spring-JMS插件相关jar包依赖 -->
            <dependency>
                <groupId>org.apache.activemq</groupId>
                <artifactId>activemq-pool</artifactId>
                <version>${activemq-pool.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.activemq</groupId>
                <artifactId>activemq-jms-pool</artifactId>
                <version>${activemq-pool.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.solr</groupId>
                <artifactId>solr-solrj</artifactId>
                <version>${solrj.version}</version>
            </dependency>
            <dependency>
                <groupId>redis.clients</groupId>
                <artifactId>jedis</artifactId>
                <version>${jedis.version}</version>
            </dependency>
            <!-- 单元测试 -->
            <dependency>
                <groupId>junit</groupId>
                <artifactId>junit</artifactId>
                <version>${junit.version}</version>
            </dependency>
            <!-- 日志处理 -->
            <dependency>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-log4j12</artifactId>
                <version>${slf4j.version}</version>
            </dependency>
            <!-- Mybatis -->
            <dependency>
                <groupId>org.mybatis</groupId>
                <artifactId>mybatis</artifactId>
                <version>${mybatis.version}</version>
            </dependency>
            <dependency>
                <groupId>org.mybatis</groupId>
                <artifactId>mybatis-spring</artifactId>
                <version>${mybatis.spring.version}</version>
            </dependency>
            <!-- MySql -->
            <dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
                <version>${mysql.version}</version>
            </dependency>
            <!-- 连接池 -->
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>druid</artifactId>
                <version>${druid.version}</version>
            </dependency>
            <!-- Spring -->
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-context</artifactId>
                <version>${spring.version}</version>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-beans</artifactId>
                <version>${spring.version}</version>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-webmvc</artifactId>
                <version>${spring.version}</version>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-jdbc</artifactId>
                <version>${spring.version}</version>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-aspects</artifactId>
                <version>${spring.version}</version>
            </dependency>
            <dependency>
                <groupId>org.springframework</groupId>
                <artifactId>spring-aop</artifactId>
                <version>${spring.version}</version>
            </dependency>
            <!-- JSP相关 -->
            <dependency>
                <groupId>jstl</groupId>
                <artifactId>jstl</artifactId>
                <version>${jstl.version}</version>
            </dependency>
            <dependency>
                <groupId>javax.servlet</groupId>
                <artifactId>servlet-api</artifactId>
                <version>${servlet-api.version}</version>
                <scope>provided</scope>
            </dependency>
            <dependency>
                <groupId>javax.servlet</groupId>
                <artifactId>jsp-api</artifactId>
                <version>${jsp-api.version}</version>
                <scope>provided</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>
    <build>
        <resources>
            <resource>
                <directory>src/main/java</directory>
                <includes>
                    <include>**/*.xml</include>
                </includes>
            </resource>
            <resource>
                <directory>src/main/resources</directory>
                <includes>
                    <include>**/*.xml</include>
                    <include>**/*.properties</include>
                </includes>
            </resource>
        </resources>
        <!-- tomcat插件，由于子项目不一定每个都是web项目，所以该插件只是声明，并未开启 -->
        <pluginManagement>
            <plugins>
                <!-- 配置Tomcat插件 -->
                <plugin>
                    <groupId>org.apache.tomcat.maven</groupId>
                    <artifactId>tomcat7-maven-plugin</artifactId>
                    <version>${tomcat.version}</version>
                </plugin> 
                <plugin>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <configuration>
                        <source>11</source>
                        <target>11</target>
                    </configuration>
                </plugin>
            </plugins>
        </pluginManagement>
    </build>
</project>

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:jms="http://www.springframework.org/schema/jms"
    xmlns:context="http://www.springframework.org/schema/context"
    xmlns:amq="http://activemq.apache.org/schema/core"
    xsi:schemaLocation="
        http://www.springframework.org/schema/beans 
        http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/jms
        http://www.springframework.org/schema/jms/spring-jms.xsd
        http://activemq.apache.org/schema/core
        http://activemq.apache.org/schema/core/activemq-core.xsd
        http://www.springframework.org/schema/context
        http://www.springframework.org/schema/context/spring-context.xsd">
    <context:property-placeholder location="classpath:jms.properties" />
    <!-- 需要创建一个连接工厂,连接ActiveMQ. ActiveMQConnectionFactory. 需要依赖ActiveMQ提供的amq标签 -->
    <!-- amq:connectionFactory 是bean标签的子标签, 会在spring容器中创建一个bean对象. 可以为对象命名. 
        类似: <bean id="" class="ActiveMQConnectionFactory"></bean> -->
    <amq:connectionFactory id="amqConnectionFactory" userName="${USERNAME}" password="${PASSWORD}" brokerURL="${BROKERURL}" />
    <!-- 配置池化的ConnectionFactory。 为连接ActiveMQ的connectionFactory提供连接池 -->
    <bean id="pooledConnectionFactory" class="org.apache.activemq.jms.pool.PooledConnectionFactory">
        <property name="connectionFactory" ref="amqConnectionFactory"></property>   
        <property name="maxConnections" value="10"></property>
    </bean>
    <!-- spring管理JMS相关代码的时候,必须依赖jms标签库. spring-jms提供的标签库. -->
    <!-- 定义Spring-JMS中的连接工厂对象 CachingConnectionFactory - spring框架提供的连接工厂对象. 
        不能真正的访问MOM容器. 类似一个工厂的代理对象. 需要提供一个真实工厂,实现MOM容器的连接访问. -->
    <!-- 配置有缓存的ConnectionFactory，session的缓存大小可定制。 -->
    <bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
        <property name="targetConnectionFactory" ref="pooledConnectionFactory"></property>
        <property name="sessionCacheSize" value="3"></property>
    </bean>
    <!-- JmsTemplate配置 -->
    <bean id="template" class="org.springframework.jms.core.JmsTemplate">
        <!-- 给定连接工厂, 必须是spring创建的连接工厂. -->
        <property name="connectionFactory" ref="connectionFactory"></property>
        <!-- 可选 - 默认目的地命名 -->
        <property name="defaultDestinationName" value="test-spring"></property>
    </bean>
</beans>

USERNAME=admin
PASSWORD=admin
BROKERURL=tcp://192.168.40.128:61616

@Service
public class UserServiveImpl implements UserService {   
    @Autowired
    private JmsTemplate jmsTemplate;
    @Override
    public void addUser(final User user) {
        //可以通过JmsTemplate对象设置queue
        //jmsTemplate.setDefaultDestinationName("aaa");
        jmsTemplate.send(new MessageCreator() {
            @Override
            public Message createMessage(Session session) throws JMSException {
                ObjectMessage message = session.createObjectMessage(user);
                return message;
}

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:jms="http://www.springframework.org/schema/jms"
    xmlns:context="http://www.springframework.org/schema/context"
    xmlns:amq="http://activemq.apache.org/schema/core"
    xsi:schemaLocation="
        http://www.springframework.org/schema/beans 
        http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/jms
        http://www.springframework.org/schema/jms/spring-jms.xsd
        http://activemq.apache.org/schema/core
        http://activemq.apache.org/schema/core/activemq-core.xsd
        http://www.springframework.org/schema/context
        http://www.springframework.org/schema/context/spring-context.xsd">
    <context:property-placeholder location="classpath:jms.properties"/>
    <!-- 需要创建一个连接工厂,连接ActiveMQ. ActiveMQConnectionFactory. 需要依赖ActiveMQ提供的amq标签 -->
    <!-- amq:connectionFactory 是bean标签的子标签, 会在spring容器中创建一个bean对象. 可以为对象命名. 
        类似: <bean id="" class="ActiveMQConnectionFactory"></bean> -->
    <amq:connectionFactory id="amqConnectionFactory" userName="${USERNAME}" password="${PASSWORD}" brokerURL="${BROKERURL}" 
     trustAllPackages="true"/>
    <!-- 配置池化的ConnectionFactory。 为连接ActiveMQ的connectionFactory提供连接池 -->
    <bean id="pooledConnectionFactory" class="org.apache.activemq.jms.pool.PooledConnectionFactory">
        <property name="connectionFactory" ref="amqConnectionFactory"></property>   
        <property name="maxConnections" value="10"></property>
    </bean>
    <!-- spring管理JMS相关代码的时候,必须依赖jms标签库. spring-jms提供的标签库. -->
    <!-- 定义Spring-JMS中的连接工厂对象 CachingConnectionFactory - spring框架提供的连接工厂对象. 
        不能真正的访问MOM容器. 类似一个工厂的代理对象. 需要提供一个真实工厂,实现MOM容器的连接访问. -->
    <!-- 配置有缓存的ConnectionFactory，session的缓存大小可定制。 -->
    <bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
        <property name="targetConnectionFactory" ref="pooledConnectionFactory"></property>
        <property name="sessionCacheSize" value="3"></property>
        <!-- <property name="trustAllPackages" value="true"></property> -->
    </bean>
    <!-- 注册监听器 -->
    <!-- 开始注册监听. 
        需要的参数有:
            acknowledge - 消息确认机制
            container-type - 容器类型 default|simple
            simple:SimpleMessageListenerContainer最简单的消息监听器容器，只能处理固定数量的JMS会话，且不支持事务。
            default:DefaultMessageListenerContainer是一个用于异步消息监听器容器 ，且支持事务         
            destination-type - 目的地类型. 使用队列作为目的地.
            connection-factory - 连接工厂, spring-jms使用的连接工厂,必须是spring自主创建的
            不能使用三方工具创建的工程. 如: ActiveMQConnectionFactory.
    <jms:listener-container acknowledge="auto" container-type="default" destination-type="queue" connection-factory="connectionFactory" >
        <!-- 在监听器容器中注册某监听器对象.
            destination - 设置目的地命名
            ref - 指定监听器对象
        <jms:listener destination="test-spring" ref="queueMessageListener"/>
    </jms:listener-container>
</beans>

@Component
public class QueueMessageListener implements MessageListener{
    @Autowired
    private UserService userServiceImpl;
    @Override
    public void onMessage(Message message) {
        try {
            User user=(User)((ObjectMessage)message).getObject();