java 中多线程的优化

左手的ㄟ右手 2023-10-13 19:24 16阅读 0赞

如果你想在Java中实现消息队列的循环处理，可以考虑以下几种方法：

1.  使用无限循环：在消费者处理消息的逻辑中，使用一个无限循环来不断地从消息队列中取出消息进行处理。当队列为空时，消费者线程会阻塞等待新的消息到达。这样可以实现循环处理消息的效果。

import java.util.concurrent.*;
    
    class Message {
        
        private String content;
    
        public Message(String content) {
        
            this.content = content;
        }
    
        public String getContent() {
        
            return content;
        }
    }
    
    public class QueueExample {
        
        public static void main(String[] args) {
        
            // 创建消息队列
            BlockingQueue<Message> messageQueue = new ArrayBlockingQueue<>(10);
    
            // 消息生产者
            Runnable producer = () -> {
        
                try {
        
                    // 模拟消息生成过程
                    for (int i = 1; i <= 10; i++) {
        
                        String content = "Data " + i;
                        Message message = new Message(content);
                        messageQueue.put(message); // 将消息添加到队列
                        System.out.println("Produced: " + content);
                        Thread.sleep(1000);
                    }
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            };
    
            // 消息消费者
            Runnable consumer = () -> {
        
                try {
        
                    // 无限循环处理消息
                    while (true) {
        
                        Message message = messageQueue.take(); // 阻塞获取消息
                        String data = message.getContent();
                        System.out.println("Consumed: " + data);
                        // 处理数据
                        process(data);
                    }
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            };
    
            // 启动消息生产者和消费者线程
            new Thread(producer).start();
            new Thread(consumer).start();
        }
    
        private static void process(String data) {
        
            // 处理数据的逻辑
            System.out.println("Processing: " + data);
        }
    }

1.  使用循环计数器：在消息消费者的处理逻辑中引入一个循环计数器，控制消息的循环处理次数。当消费者线程处理完固定次数的消息后，可以选择终止或重置计数器，继续从队列中处理新的消息。

import java.util.concurrent.*;
    
    class Message {
        
        private String content;
    
        public Message(String content) {
        
            this.content = content;
        }
    
        public String getContent() {
        
            return content;
        }
    }
    
    public class QueueExample {
        
        public static void main(String[] args) {
        
            // 创建消息队列
            BlockingQueue<Message> messageQueue = new ArrayBlockingQueue<>(10);
    
            // 消息生产者
            Runnable producer = () -> {
        
                try {
        
                    // 模拟消息生成过程
                    for (int i = 1; i <= 10; i++) {
        
                        String content = "Data " + i;
                        Message message = new Message(content);
                        messageQueue.put(message); // 将消息添加到队列
                        System.out.println("Produced: " + content);
                        Thread.sleep(1000);
                    }
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            };
    
            // 消息消费者
            Runnable consumer = () -> {
        
                int count = 0;
                int maxCount = 3; // 每个消息循环处理的最大次数
                try {
        
                    // 从队列中获取消息并处理
                    while (true) {
        
                        Message message = messageQueue.take(); // 阻塞获取消息
                        String data = message.getContent();
                        System.out.println("Consumed: " + data);
                        // 处理数据
                        process(data);
                        count++;
                        if (count == maxCount) {
        
                            break; // 达到最大处理次数，退出循环
                        }
                        // 重置计数器
                        if (!messageQueue.isEmpty()) {
        
                            count = 0;
                        }
                    }
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            };
    
            // 启动消息生产者和消费者线程
            new Thread(producer).start();
            new Thread(consumer).start();
        }
    
        private static void process(String data) {
        
            // 处理数据的逻辑
            System.out.println("Processing: " + data);
        }
    }

这两种