ActiveMQ实际应用实例

约定不等于承诺〃 2022-05-28 08:25 280阅读 0赞

## 前提 ##

本例子是模拟下订单的，结合 Spring、SpringMVC 整合的 ActiveMQ。

*  MySQL 表结构及数据

CREATE TABLE `tb_cart` ( `id` int(11) NOT NULL AUTO_INCREMENT, `goods_id` int(11) NOT NULL COMMENT '商品 id', `goods_quantity` int(11) NOT NULL COMMENT '商品数量', `goods_price` int(11) NOT NULL COMMENT '价格', `customer_id` int(11) NOT NULL COMMENT '用户 id', PRIMARY KEY (`id`) ) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;
    INSERT INTO `tb_cart` VALUES ('1', '1', '1', '10', '1');
    CREATE TABLE `tb_customer` ( `id` int(11) NOT NULL AUTO_INCREMENT, `name` varchar(20) NOT NULL, PRIMARY KEY (`id`) ) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;
    INSERT INTO `tb_customer` VALUES ('1', 'yz');
    CREATE TABLE `tb_goods` ( `id` bigint(20) NOT NULL AUTO_INCREMENT, `name` varchar(20) NOT NULL COMMENT '商品名称', `price` int(11) NOT NULL COMMENT '价格', PRIMARY KEY (`id`) ) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;
    INSERT INTO `tb_goods` VALUES ('1', '旺仔小馒头', '10');
    CREATE TABLE `tb_order` ( `id` int(11) NOT NULL AUTO_INCREMENT, `customer_id` int(11) NOT NULL COMMENT '顾客 id', `create_date` date NOT NULL COMMENT '创建时间', `total_money` int(11) NOT NULL COMMENT '总价', `status` int(1) NOT NULL COMMENT '状态', PRIMARY KEY (`id`) ) ENGINE=InnoDB AUTO_INCREMENT=1131 DEFAULT CHARSET=utf8;
    CREATE TABLE `tb_stock` ( `id` int(11) NOT NULL AUTO_INCREMENT, `goods_id` int(11) NOT NULL COMMENT '商品 id', `stock` int(11) NOT NULL COMMENT '库存', PRIMARY KEY (`id`) ) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;
    INSERT INTO `tb_stock` VALUES ('1', '1', '1000');

*  jar 包

<dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-pool</artifactId>
        <version>5.9.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-all</artifactId>
        <version>5.10.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.geronimo.specs</groupId>
        <artifactId>geronimo-servlet_3.0_spec</artifactId>
        <version>1.0</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.geronimo.specs</groupId>
        <artifactId>geronimo-jms_1.1_spec</artifactId>
        <version>1.1.1</version>
    </dependency>

## demo ##

*  消息发送者，将客户 id 发送到 activemq 中

import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class QueueSender{ 
        private static ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://182.254.246.166:61616");
        public static void sendMsg(int customerId) {
            Connection connection = null;
            Session session = null;
            try {
                connection = connectionFactory.createConnection();
                connection.start();
                session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
                Destination destination = session.createQueue("order-queue");
                MessageProducer producer = session.createProducer(destination);
                producer.send(session.createTextMessage(customerId + ""));
                session.commit();
            } catch (JMSException e) {
                e.printStackTrace();
            } finally {
                try {
                    if (null != session) {
                        session.close();
                    }
                    if (null != connection) {
                        connection.close();
                    }
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        }
    }

*  消息消费者，通过客户 id 查询购物车信息、生成订单、删除购物车、减库存

import java.util.Date;
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageConsumer;
    import javax.jms.MessageListener;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import javax.servlet.ServletContextEvent;
    import javax.servlet.ServletContextListener;
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.ApplicationContext;
    import org.springframework.stereotype.Service;
    import org.springframework.web.context.support.WebApplicationContextUtils;
    import com.alibaba.fastjson.JSON;
    import com.p7.framework.entity.Order;
    import com.p7.framework.service.CartService;
    import com.p7.framework.service.OrderService;
    import com.p7.framework.service.StockService;
    
    /** * * 消息消费者实现 监听，使服务启动时，立即调用 consumeMsg 方法 */
    @Service
    public class QueueReceive implements ServletContextListener { 
        @Autowired
        private OrderService orderService;
        @Autowired
        private StockService stockService;
        @Autowired
        private CartService cartService;
        private static ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://182.254.246.166:61616");
        public void consumeMsg() {
            // 创建 10 个线程，run 方法中使用 activemq 的监听模式，使线程保持监听而不会销毁
            for (int i = 0; i < 10; i++) {
                new OrderThread(orderService, stockService, cartService, connectionFactory).start();
            }
        }
        @Override
        public void contextDestroyed(ServletContextEvent sce) {
    
        }
        @Override
        public void contextInitialized(ServletContextEvent sce) {
            ApplicationContext context = WebApplicationContextUtils.getWebApplicationContext(sce.getServletContext());
            QueueReceive bean = (QueueReceive) context.getBean("queueReceive");
            bean.consumeMsg();
        }
    }
    /** * * 模拟并发 */
    class OrderThread extends Thread {
        private OrderService orderService;
        private StockService stockService;
        private CartService cartService;
        private ConnectionFactory connectionFactory;
    
        public OrderThread(OrderService orderService, StockService stockService, CartService cartService,
                ConnectionFactory connectionFactory) {
            this.cartService = cartService;
            this.orderService = orderService;
            this.stockService = stockService;
            this.connectionFactory = connectionFactory;
        }
    
        @Override
        public void run() {
    
            System.out.println(Thread.currentThread().getName() + "------");
            Connection connection = null;
            final Session session;
    
            try {
                connection = connectionFactory.createConnection();
                connection.start();
                session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
                Destination destination = session.createQueue("order-queue");
                MessageConsumer consumer = session.createConsumer(destination);
                consumer.setMessageListener(new MessageListener() {
    
                    @Override
                    public void onMessage(Message message) {
                        try {
                            TextMessage msg = (TextMessage) message;
                            String customerId = msg.getText();
                            session.commit();
    
                            System.out.println(Thread.currentThread().getName() + "------处理时间：" + JSON.toJSON(new Date()));
                            List<Map<String, Object>> cartList = cartService.findByCustomerId(customerId);
                            Map<String, Integer> goodsNumMap = new HashMap<String, Integer>();
                            int totalMoney = 0;
                            for (Map<String, Object> map : cartList) {
                                totalMoney += Integer.parseInt(map.get("goods_price").toString())
                                        * Integer.parseInt(map.get("goods_quantity").toString());
                                goodsNumMap.put(map.get("goods_id").toString(),
                                        Integer.parseInt(map.get("goods_quantity").toString()));
                            }
                            // 创建订单
                            Order order = new Order(Integer.parseInt(customerId), new Date(), totalMoney, 1);
                            orderService.createOrder(order);
                            // 减库存，这里的锁一定要使用库存那个对象，如果不使用锁，那么并发时，减库存有问题
                            synchronized (stockService) {
                                for (String key : goodsNumMap.keySet()) {
                                    stockService.decrease(key, goodsNumMap.get(key));
                                }
                            }
                            // 删除购物车，这里是为了使用同一条数据模拟并发，因此不删除购物车数据
                            // cartService.deleteByCustomerId(customerId);
                        } catch (JMSException e) {
                        }
                    }
                });
            } catch (JMSException e) {
            } finally {
            }
        }
    }

*  web.xml

<context-param>
        <param-name>contextConfigLocation</param-name>
        <param-value>classpath:config/spring/applicationContext.xml</param-value>
    </context-param>
    <listener>
        <listener-class>org.springframework.web.context.ContextLoaderListener</listener-class>
    </listener>
    
    <listener>
        <listener-class>com.p7.framework.ordermq.QueueReceive</listener-class>
    </listener>

*  购物车 code

/** 接口 */
    import java.util.List;
    import java.util.Map;
    public interface CartService { 
        /** * 根据客户 id 查询购物车信息 * @param customerId * @return */
        List<Map<String, Object>> findByCustomerId(String customerId);
        /** * 根据客户 id 删除购物车信息 * @param customerId */
        void deleteByCustomerId(String customerId);
    }
    
    /** 接口实现 */
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
    import org.springframework.stereotype.Service;
    
    import com.p7.framework.service.CartService;
    
    @Service
    public class CartServiceImpl implements CartService { 
        @Autowired
        private NamedParameterJdbcTemplate jdbcTemplate;
        @Override
        public List<Map<String, Object>> findByCustomerId(String customerId) {
            Map<String, Object> params = new HashMap<String, Object>();
            params.put("customerId", customerId);
            List<Map<String, Object>> resultList = jdbcTemplate
                    .queryForList("select * from tb_cart where customer_id=:customerId", params);
            return resultList;
        }
        @Override
        public void deleteByCustomerId(String customerId) {
            Map<String, Object> params = new HashMap<String, Object>();
            params.put("customerId", customerId);
            jdbcTemplate.update("delete from tb_cart where customer_id=:customerId ", params);
        }
    }

*  订单 code

/** 接口 */
    import com.p7.framework.entity.Order;
    public interface OrderService { 
        void createOrder(Order order);
    }
    
    /** 接口实现 */
    import java.util.Map;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
    import org.springframework.stereotype.Service;
    import com.p7.framework.entity.Order;
    import com.p7.framework.service.OrderService;
    @Service
    public class OrderServiceImpl implements OrderService { 
        @Autowired
        private NamedParameterJdbcTemplate jdbcTemplate;
        @Override
        public void createOrder(Order order) {
            Map<String, Object> map = order.toMap();
            jdbcTemplate.update(
                    "insert into tb_order(customer_id,create_date,total_money,status) values(:customer_id , :create_date , :total_money , :status )",
                    map);
        }
    }

*  库存 code

/** 接口 */
    package com.p7.framework.service;
    public interface StockService { 
        void decrease(String goodsId, int num);
    }
    /** 接口实现 */
    import java.util.HashMap;
    import java.util.Map;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
    import org.springframework.stereotype.Service;
    import com.p7.framework.service.StockService;
    @Service
    public class StockServiceImpl implements StockService { 
        @Autowired
        private NamedParameterJdbcTemplate jdbcTemplate;
        @Override
        public void decrease(String goodsId, int num) {
            Map<String, Object> params = new HashMap<String, Object>();
            params.put("goodsId", goodsId);
            Map<String, Object> resultMap = jdbcTemplate.queryForMap("select stock from tb_stock where goods_id=:goodsId ",
                    params);
            Integer stock = Integer.parseInt(resultMap.get("stock").toString());
            params.put("stock", stock - num);
            jdbcTemplate.update("update tb_stock set stock=:stock where goods_id=:goodsId ", params);
        }
    }