java - 在Web应用程序中实现生产者-消费者的最佳方式

Question

这是我的场景: 每一分钟都有一个线程检查数据库以查找某些项目。 当找到项目时，它们将被传递给另一个线程(消费者)。

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

请帮助我了解实现我的场景的正确方法是什么。

Answer 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();
            }
        }
    }

}