在 Web 应用中实现生产者-消费者的最佳方式答案

【问题标题】：The best way to implement producer-consumer in web app在 Web 应用中实现生产者-消费者的最佳方式
【发布时间】：2016-07-01 19:18:41
【问题描述】：

这是我的场景：每分钟一个线程检查数据库以查找某些项目。找到项目后，将它们传递给另一个线程（消费者）。

我在我的应用程序中使用了 spring，但这并不意味着我必须使用 spring 的类，对吗？所以，现在我完全迷失在ScheduledExecutorService (java)、ExecutorService (java)、TaskExecutor (spring)、TaskScheduler (spring)、@Scheduled (spring) 的树林中。

请帮助我了解实现我的方案的正确方法。

【问题讨论】：

我有一个类似的情况，我为生产者创建线程，为消费者创建大小为 4 的 FixedThreadPool。
对于如此广泛的问题没有“正确的方法”。如果您有弹簧，请使用它。

标签： java spring multithreading web-applications concurrency

【解决方案1】：

如果我理解您的问题，您使用的是 Shared Database Pattern，因为 many reason 有点不鼓励并用作最后的资源。

如果您希望多个应用程序进行通信并解耦，则应使用Messaging（例如：Spring Cloud Stream）。

无论如何，如果您需要拥有Shared Database，您可能需要Listen / Notify 提供的内容。

来自https://jdbc.postgresql.org/documentation/81/listennotify.html：

import java.sql.*;

public class NotificationTest {

    public static void main(String args[]) throws Exception {
        Class.forName("org.postgresql.Driver");
        String url = "jdbc:postgresql://localhost:5432/test";

        // Create two distinct connections, one for the notifier
        // and another for the listener to show the communication
        // works across connections although this example would
        // work fine with just one connection.
        Connection lConn = DriverManager.getConnection(url,"test","");
        Connection nConn = DriverManager.getConnection(url,"test","");

        // Create two threads, one to issue notifications and
        // the other to receive them.
        Listener listener = new Listener(lConn);
        Notifier notifier = new Notifier(nConn);
        listener.start();
        notifier.start();
    }

}

监听器

class Listener extends Thread {

    private Connection conn;
    private org.postgresql.PGConnection pgconn;

    Listener(Connection conn) throws SQLException {
        this.conn = conn;
        this.pgconn = (org.postgresql.PGConnection)conn;
        Statement stmt = conn.createStatement();
        stmt.execute("LISTEN mymessage");
        stmt.close();
    }

    public void run() {
        while (true) {
            try {
                // issue a dummy query to contact the backend
                // and receive any pending notifications.
                Statement stmt = conn.createStatement();
                ResultSet rs = stmt.executeQuery("SELECT 1");
                rs.close();
                stmt.close();

                org.postgresql.PGNotification notifications[] = pgconn.getNotifications();
                if (notifications != null) {
                    for (int i=0; i<notifications.length; i++) {
                        System.out.println("Got notification: " + notifications[i].getName());
                    }
                }

                // wait a while before checking again for new
                // notifications
                Thread.sleep(500);
            } catch (SQLException sqle) {
                sqle.printStackTrace();
            } catch (InterruptedException ie) {
                ie.printStackTrace();
            }
        }
    }

}

通知器

class Notifier extends Thread {

    private Connection conn;

    public Notifier(Connection conn) {
        this.conn = conn;
    }

    public void run() {
        while (true) {
            try {
                Statement stmt = conn.createStatement();
                stmt.execute("NOTIFY mymessage");
                stmt.close();
                Thread.sleep(2000);
            } catch (SQLException sqle) {
                sqle.printStackTrace();
            } catch (InterruptedException ie) {
                ie.printStackTrace();
            }
        }
    }

}

【讨论】：

【解决方案2】：

您可以使用石英调度程序而不是使用 ScheduledExecutorService，该调度程序用于以特定的时间间隔安排一些作业，在您的情况下是每分钟一次。它可以很容易地与弹簧集成。 Cron 表达式用于指定计划的时间。您可以编写逻辑来检查数据库以查找扩展 QuartzJobBean 类的类中的某些项目。看： https://examples.javacodegeeks.com/enterprise-java/quartz/spring-quartz-scheduler-example/

【讨论】：