rabbitmq消息中间件的案例

﹏ヽ暗。殇╰゛Y 2022-03-25 09:41 233阅读 0赞

GitHub地址：[https://github.com/hsowan/rabbitmq-demo][https_github.com_hsowan_rabbitmq-demo]

This demo will involve RabbitMQ and SpringAMQP.

### Application ###

*  SecKill simulation
 *  Reliable delivery

### SecKill ###

Now, there is only one goods which is `MacBook Pro` and one hundred users want to get it, ok,  
the only one `MacBook Pro` will be free after n minutes and all users can `SecKill` it. Let\`s  
realize it.

> Using ThreadPoolExecutor to fake 100 users

/** * int corePoolSize, * int maximumPoolSize, * long keepAliveTime, * java.util.concurrent.TimeUnit unit, * java.util.concurrent.BlockingQueue<Runnable> workQueue */
    private ThreadPoolExecutor threadPoolExecutor = 
            new ThreadPoolExecutor(100, 1000, 200, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(10));

> Expose problem in SecKill situation

// Create 100 thread waiting for the SecKill
    for (int i = 1; i <= 100; i++) { 
        threadPoolExecutor.execute(() -> { 
            while (true) { 
                if (System.currentTimeMillis() >= startTime) { 
    
                    Timestamp createAndUpdateTime = new Timestamp(System.currentTimeMillis());
    
                    String messageId = UUID.randomUUID().toString();
    
                    // Create an order
                    orderService.saveOrUpdate(new Order("order-macbookpro", OrderConstants.CREATING, createAndUpdateTime, createAndUpdateTime, messageId));
    
                    // If count > 0, switch the status of an order to DONE
                    Goods goods = goodsService.getById((long) 1);
                    if (goods.getCount() > 0){ 
    
                        Order order = orderService.getByMessageid(messageId);
                        if (order != null){ 
                            order.setStatus(OrderConstants.DONE);
                            order.setUpdateTime(new Timestamp(System.currentTimeMillis()));
                            orderService.saveOrUpdate(order);
                        }
    
                        goods.setCount(goods.getCount() - 1);
                        goodsService.saveOrUpdate(goods);
                    }
                    break;
                }
            }
        });
    }

What will happen?

mysql> select count(1) from test_order where status = 1;
    +----------+
    | count(1) |
    +----------+
    |      100 |
    +----------+
    1 row in set (0.00 sec)

All users got the only one `MacBook Pro`!!!

> Solve it with using RabbitMQ

This is the OrderProducer that sends messages.

public void sendMessage(Order order){ 
        Long time = System.currentTimeMillis();
        Timestamp createAndUpdateTime = new Timestamp(time);
        Timestamp nextRetry = new Timestamp(time + 60000 * MessageConstants.INTERVAL);
    
        Message message = new Message(JSON.toJSONString(order), MessageConstants.SENDING, 0, nextRetry, createAndUpdateTime, createAndUpdateTime);
    
        rabbitTemplate.setConfirmCallback((correlationData, b, s) -> { 
            if (b){ 
                System.out.println("ack");
            }else { 
                System.out.println("nack");
            }
        });
        CorrelationData correlationData = new CorrelationData();
        rabbitTemplate.convertAndSend("order-exchange", "order.secKill", JSON.toJSONString(message), correlationData);
    }

This is the OrderConsumer that mostly waits to receive messages.

@RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "order-queue", durable = "true"),
            exchange = @Exchange(name = "order-exchange", type = "topic"),
            key = "order.*"
    ))
    @RabbitHandler
    public void onMessage(@Payload String message, @Headers Map<String, Object> headers, Channel channel) throws IOException { 
        Message m = JSON.parseObject(message, Message.class);
        System.out.println(m.toString());
    
        String messageId = JSON.parseObject(m.getBody(), Order.class).getMessageId();
    
        Goods goods = goodsService.getById((long) 1);
        Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
        int count = goods.getCount();
        if (count > 0){ 
            goods.setCount(count - 1);
            goodsService.saveOrUpdate(goods);
    
            Order order = orderService.getByMessageid(messageId);
            order.setStatus(OrderConstants.DONE);
            order.setUpdateTime(new Timestamp(System.currentTimeMillis()));
            orderService.saveOrUpdate(order);
    
        }
        channel.basicAck(deliveryTag, false);
    }

Make sure that there is a single RabbitMQ server running background.  
Then run the consumer (run `Application.java` in client module).  
Finally, run the producer (run the `testUsingRabbitMQ` test of the `ApplicationTests.java` in server module).

mysql> select count(1) from test_order where status = 1;
    +----------+
    | count(1) |
    +----------+
    |        1 |
    +----------+
    1 row in set (0.00 sec)

Cool!

### AMQP ###

> Core Concepts

*  Server
 *  Connection
 *  Channel(like session in jdbc)
 *  Message(properties & body)
 *  Virtual host
 *  Exchange
 *  Binding
 *  Routing key
 *  Queue

> AMQ Model

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2tlbjE1ODMwOTY2ODM_size_16_color_FFFFFF_t_70]

> Message Flow

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2tlbjE1ODMwOTY2ODM_size_16_color_FFFFFF_t_70 1]

### Tutorial ###

> [Hello World][]

*  message broker
 *  producer
 *  consumer
 *  queue

![在这里插入图片描述][2019012004440141.png]

> [Work Queues][]

*  multiple workers
 *  round-robin dispatching
 *  message acknowledgment (ack)
 *  message durability
 *  fair dispatch
 *  prefetch
 *  qos

![在这里插入图片描述][20190120044413651.png]

> [Publish/Subscribe][Publish_Subscribe]

The core idea in the messaging model in RabbitMQ is that the producer never sends any messages directly to a queue.  
Actually, quite often the producer doesn’t even know if a message will be delivered to any queue at all.

*  exchange
 *  exchange type (direct, topic, headers, fanout)
 *  broadcast
 *  temporary queue (non-durable, exclusive, autodelete, random queue name)
 *  binding
 *  routingKey (its value is ignored for fanout exchanges)

![在这里插入图片描述][20190120044439395.png]

![在这里插入图片描述][20190120044451860.png]

> [Routing][] (Direct)

A binding is a relationship between an exchange and a queue.  
This can be simply read as: the queue is interested in messages from this exchange.

The routing algorithm behind a direct exchange is simple - a message goes to the queues whose binding key exactly matches the routing key of the message.

*  binding key (routingKey)

![在这里插入图片描述][2019012004450828.png]

> [Topics][]

Messages sent to a topic exchange can’t have an arbitrary routing\_key - it must be a list of words, `delimited by dots`.  
The words can be anything, but usually they specify some features connected to the message.  
A few valid routing key examples: “stock.usd.nyse”, “nyse.vmw”, “quick.orange.rabbit”.  
There can be as many words in the routing key as you like, up to the limit of `255 bytes`.

*  `* (star)` can substitute for exactly one word.
 *  `# (hash)` can substitute for zero or more words.

> [Remote procedure call (RPC)][Remote procedure call _RPC]

*  Callback queue
 *  Correlation Id

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2tlbjE1ODMwOTY2ODM_size_16_color_FFFFFF_t_70 2]

### Docker ###

$ docker run -d \
    --name first-rabbit \
    --hostname rabbitmq-demo \
    -p 5672:5672 \
    -p 8080:15672 \
    -e RABBITMQ_DEFAULT_USER=user \
    -e RABBITMQ_DEFAULT_PASS=password \
    rabbitmq:3-management

You can then go to [http://localhost:8080][http_localhost_8080] or [http://host-ip:8080][http_host-ip_8080] in a browser  
and use `user` / `password` to gain access to the management console.

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2tlbjE1ODMwOTY2ODM_size_16_color_FFFFFF_t_70 3]

### Documents ###

*  [RabbitMQ Docker Official Images][]
 *  [AMQP 0-9-1 Model Explained][]
 *  [AMQP Specification][]
 *  [Spring AMQP 2.1.3 Reference Doc][]
 *  [Java SE API Documentation][]
 *  [RabbitMQ Tutorials][]
 *  [JdbcType in MyBatis][]

### Blog ###

*  [Java Concurrent Programming: How to use ThreadPool][Java Concurrent Programming_ How to use ThreadPool]

[https_github.com_hsowan_rabbitmq-demo]: https://github.com/hsowan/rabbitmq-demo
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2tlbjE1ODMwOTY2ODM_size_16_color_FFFFFF_t_70]: /images/20220325/5813c1a87a924e3485b0d6f2ac34ac9b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2tlbjE1ODMwOTY2ODM_size_16_color_FFFFFF_t_70 1]: /images/20220325/e9570abeed2f4acb93875cd3e2a6c972.png
[Hello World]: http://www.rabbitmq.com/tutorials/tutorial-one-java.html
[2019012004440141.png]: /images/20220325/261016d9cede4a1e9a155c08359ef997.png
[Work Queues]: http://www.rabbitmq.com/tutorials/tutorial-two-java.html
[20190120044413651.png]: /images/20220325/180201317d6f4913916e9fdb2de088ae.png
[Publish_Subscribe]: http://www.rabbitmq.com/tutorials/tutorial-three-java.html
[20190120044439395.png]: /images/20220325/565118a9b4dc4f5aaf2f16c956c34051.png
[20190120044451860.png]: /images/20220325/643185ea2ae140558b54fd6992c9bfd3.png
[Routing]: http://www.rabbitmq.com/tutorials/tutorial-four-java.html
[2019012004450828.png]: /images/20220325/c1b18a1da8da46d88e50fb519c49f0e0.png
[Topics]: http://www.rabbitmq.com/tutorials/tutorial-five-java.html
[Remote procedure call _RPC]: http://www.rabbitmq.com/tutorials/tutorial-six-java.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2tlbjE1ODMwOTY2ODM_size_16_color_FFFFFF_t_70 2]: /images/20220325/cb3da9e80c5b497f8441d7220c29aa06.png
[http_localhost_8080]: http://localhost:8080
[http_host-ip_8080]: http://host-ip:8080
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2tlbjE1ODMwOTY2ODM_size_16_color_FFFFFF_t_70 3]: /images/20220325/8888591d00d04e3e9559513c524b97f6.png
[RabbitMQ Docker Official Images]: https://hub.docker.com/_/rabbitmq/
[AMQP 0-9-1 Model Explained]: http://www.rabbitmq.com/tutorials/amqp-concepts.html
[AMQP Specification]: http://www.rabbitmq.com/resources/specs/amqp0-9-1.pdf
[Spring AMQP 2.1.3 Reference Doc]: https://docs.spring.io/spring-amqp/docs/2.1.3.RELEASE/reference/html/
[Java SE API Documentation]: https://docs.oracle.com/javase/8/docs/api/index.html
[RabbitMQ Tutorials]: http://www.rabbitmq.com/getstarted.html
[JdbcType in MyBatis]: http://www.mybatis.org/mybatis-3/apidocs/reference/org/apache/ibatis/type/JdbcType.html
[Java Concurrent Programming_ How to use ThreadPool]: https://www.cnblogs.com/dolphin0520/p/3932921.html