【项目实战典型案例】16.消息队列作用和意义

2023-07-27 111

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： 【项目实战典型案例】16.消息队列作用和意义

一：背景介绍

二：消息队列

消息队列简介

MQ全程为Message Queue，消息队列（MQ）是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息来通信，而无需专用连接来链接它们。
消息传递指的是程序之间通过在消息中发送数据进行通信，而不是通过直接调用彼此来通信，直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求

消息队列中间件是分布式系统中重要的组件，主要解决应用耦合、异步消息、流量削峰等问题。

解耦

譬如签到送积分，签到和送积分是两个操作。签到产生了很重要的数据，它可以把消息发送到MQ，然后积分系统需要该数据，从MQ中直接获取即可。这样签到系统就做到了和积分系统解耦，不必担心积分系统挂了怎么办，是不是需要重试等，而这些都可以在积分系统内部自己实现，再者，如果以后另外一套系统也需要该签到数据，直接从MQ中获取即可，实际上与签单系统已无关系。

异步

当做到解耦后，实现异步就是自然而然的事情，如果签到只需要1ms，而送积分，或者其他操作需要500ms，那不可能等所有操作完成之后再去返回数据给用户，这样就做到了异步。

流量削峰

削峰是指当并发访问高峰期，通过MQ达到限流的目的，从而减少对数据库MySQL的压力

原理

1.ArrayBockingQueue：

ArrayBlockingQueue是一个阻塞式的队列，继承自AbstractBlockingQueue，底层以数组的形式保存数据(实际上可看作一个循环数组)

ArrayBockingQueue使用场景：

1.先进先出队列：头是先进队的元素，尾是后进队的元素

2.有界队列：初始化时指定的容量，就是队列最大的容量，不会出现扩容，容量满，则阻塞进队操作；容量空，则阻塞出队操作

3.队列不支持空元素

阻塞式队列方法的四种形式：

2.Socket

创建一个Socket类的实例，用它来发送和接收字节流，发送时调用getOutputStream方法获取一个java.io.OutputStream对象，接收远程对象发送来的信息可以调用getInputStream方法来返回一个java.io.InputStream对象

3.SeverSocket

ServerSocket与Socket不同，ServerSocket是等待客户端的请求，一旦获得一个连接请求，就创建一个Socket示例来与客户端进行通信。

4.Java IO操作——BufferedReader

可以接收任意长度的数据，并且避免乱码的产生

5.java.io.PrintWriter

输出流、字符打印流

三：实现过程

解耦和异步

消息处理中心 Broker

public class Broker {
    private final static int MAX_SIZE = 3;
    private static ArrayBlockingQueue<String> messageQueue = new ArrayBlockingQueue<>(MAX_SIZE);
    public static void produce(String msg){
        if(messageQueue.offer(msg)){
            System.out.println("已成功向消息处理中心发送消息： " + msg + ",当前缓存的消息数量是："+ messageQueue.size());
        } else{
            System.out.println("消息处理中心内暂存的消息达到最大负荷，不能继续放入消息！");
        }
        System.out.println("-----------------------------");
    }
    public static String consume(){
        String msg = messageQueue.poll();
        if(msg != null){
            System.out.println("已经消费的消息：" + msg + ",当前暂存消息的数量是：" + messageQueue.size());
        } else {
            System.out.println("消息处理中心内没有可供消费的消息！");
        }
        System.out.println("-----------------------------");
        return msg;
    }
}

BrokerSever用来提供Broker类得对外服务，BrokerSever类实现Runnable接口，实现run方法。用new Thread(Runnable target).start()方法来启动

public class BrokerSever implements Runnable{
    public static int SERVICE_PORT = 9999;
    private final Socket socket;
    public BrokerSever(Socket socket){
        this.socket = socket;
    }
    @Override
    public void run() {
        try(
                BufferedReader in = new BufferedReader(new InputStreamReader(
                        socket.getInputStream()));
                PrintWriter out = new PrintWriter(socket.getOutputStream()))
        {
            while (true){
                String str = in.readLine();
                if (str == null){
                    continue;
                }
                System.out.println("接收到原始数据： " + str);
                if (str.equals("CONSUME")){
                    String message = Broker.consume();
                    out.println(message);
                    out.flush();
                }else {
                    Broker.produce(str);
                }
            }
        } catch (Exception e){
            e.printStackTrace();
        }
    }
    public static void main(String[] args) throws Exception{
        ServerSocket server = new ServerSocket(SERVICE_PORT);
        while(true){
            BrokerSever brokerServer = new BrokerSever(server.accept());
            new Thread(brokerServer).start();
        }
    }
}

ProduceClient消息生产者

public class ProduceClient {
    public static void main(String[] args) throws Exception{
        MyClient client = new MyClient();
        client.produce("hello World.");
    }
}

ConsumeClient消息消费者

public class ConsumeClient {
    public static void main(String[] args) throws Exception{
        MyClient client = new MyClient();
        String message = client.consume();
        System.out.println("获得的消息为： " + message);
    }
}

MyClient与消息服务器进行通信

public class MyClient {
    public static void produce(String message) throws Exception{
        Socket socket = new Socket(InetAddress.getLocalHost(),BrokerSever.SERVICE_PORT);
        try(
                PrintWriter out = new PrintWriter(socket.getOutputStream())
        ){
            out.println(message);
            out.flush();
        }
    }
    public static String consume() throws Exception{
        Socket socket = new Socket(InetAddress.getLocalHost(),BrokerSever.SERVICE_PORT);
        try(
                BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
                PrintWriter out = new PrintWriter(socket.getOutputStream())
        ){
            out.println("CONSUME");
            out.flush();
            String message = in.readLine();
            return message;
        }
    }
}

流量削峰

定义一个消息生产者

@Test
public void test() throws Exception {
    for (int i = 0; i < 1000 ; i++) {
        rabbitTemplate.convertAndSend("test-queue ",  "消息发送);
    }
    Thread.sleep(1000 * 1000);
}

使用@RabbitListener注解定义一个消息消费者

@Component
@RabbitListener(queuesToDeclare = @Queue(name = "test-queue"))
public class Consumer {
    private int count = 0;
    @RabbitHandler
    public void receive(String msg, Channel channel, Message message) throws IOException {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            Thread.sleep(1000);
            System.out.println("=====消息处理===>");
            channel.basicAck(deliveryTag, true);
            System.out.println("current count is:" + ++count);
        } catch (Exception e) {
        }
    }
}

运行效果

Ready：待消费的消息总数

Unacked：待应答的消息总数。

Total：总数 Ready+Unacked

四：总结

大家可以参考一下这篇博客：消息队列的作用和整体介绍

【项目实战典型案例】16.消息队列作用和意义

一：背景介绍

二：消息队列

消息队列简介

解耦

异步

流量削峰

原理

1.ArrayBockingQueue：

2.Socket

3.SeverSocket

4.Java IO操作——BufferedReader

5.java.io.PrintWriter

三：实现过程

解耦和异步

流量削峰

四：总结

热门文章

最新文章

相关课程

相关电子书

相关实验场景