Нотификация в PostgreSQL

Здравствуй, дорогой друг!
Многие из вас, кто уже работает разработчиком, конечно знают, что база данных - это фундамент абсолютного большинства создаваемых систем. Это средоточие информации, сердце практически любой системы.
Но, возможно, вы не знаете, что база данных - это не только ценный мех хранилище, но и отличный мессенджер =)
В этой статье мы обсудим, как обмениваться сообщениями в СУБД PostgreSQL.
Поехали!

Немного теории

PostgreSQL - это замечательная свободная и бесплатная СУБД, которая, наряду с MySQL, является одной из самых популярных свободных СУБД в мире.

Для того, чтобы обмениваться сообщениями в PostgreSQL, нам понадобятся следующие SQL-команды:

-- Запуск сеанса получения сообщений
LISTEN канал

-- Остановка получения сообщений
UNLISTEN канал

-- Отправка сообщения
NOTIFY канал, 'сообщение'

Эти три команды позволяют организовать обмен сообщениями между разными клиентами СУБД.
"канал" в данном случае представляет любую текстовую строку, идентифицирующую участника обмена.

Пример

Разберем небольшой пример, каким образом клиенты СУБД могут обмениваться сообщениями. Допустим некий пользователь по имени toolkas (все совпадения случайны ;)) хочет начать общаться с пользователем xxx. 

Для того, чтобы получать сообщения, toolkas должен установить соединение с СУБД и выполнить команду:

LISTEN toolkas

В то же время, пользователь xxx для отправки сообщения пользователю toolkas также должен установить соединение с СУБД и выполнить команду:

NOTIFY toolkas, 'Hello, toolkas'

В этом случае toolkas получит сообщение от xxx.
Интересный момент: если третий пользователь подключится к СУБД и выполнит команду:

LISTEN toolkas

то он так же, как и toolkas, будет получать все сообщения от xxx. Другими словами, система сообщений в PostgreSQL работает по принципу Publisher-Subscriber.
Для того, чтобы остановить прием сообщений, нужно выполнить команду:

UNLISTEN toolkas

Если вы хотите более подробно почитать об этих командах, то это можно сделать тут, тут и тут.

Работа с командами на Java

Теперь, когда мы познакомились с самими командами, посмотрим как их корректно вызывать с помощью языка программирования Java. 

Для того, чтобы начать получать сообщения, необходимо выполнить следующий код:

try (Statement statement = connection.createStatement()) {
    statement.execute("LISTEN toolkas");
}

Чтобы остановить прием сообщений, необходимо выполнить код:

try (Statement statement = connection.createStatement()) {
    statement.execute("UNLISTEN toolkas");
}

И, наконец, чтобы отправить сообщение, нужно выполнить код:

try (Statement statement = connection.createStatement()) {
    statement.execute("NOTIFY toolkas, 'Hello, world'");
}

Теперь, когда мы вооружились необходимыми знаниями, мы можем написать полноценную консольную программу для обмена сообщениями. Поскольку эта программа однозначно лучше всех существующих мессенджеров, назовем ее TheBestMessenger. Ура!

TheBestMessenger

Для начала обсудим, что именно наш мессенджер будет делать.

1. При запуске мессенджер будет спрашивать имя пользователя.

2. После ввода имени пользователя он будет запускать сеанс прослушки сообщений, посланных этому пользователю.

3. Каждое полученное сообщение мессенджер будет выводить на консоль вместе с именем отправителя.

4. Для отправки сообщений другому пользователю, необходимо в консоли набрать имя пользователя и сообщение, через пробел. 

Таким образом наш мессенджер должен запустить 2 потока: получатель и отправитель сообщений.

Сначала напишем код потока-получателя сообщений. При запуске поток сообщений должен установить соединение с СУБД и начать прослушку сообщений. При получении сообщений нужно вывести их на консоль.

Таким образом, код потока-получателя будет выглядеть примерно так:

package ru.toolkas.thebestmessenger;

import org.postgresql.PGNotification;
import org.postgresql.ds.PGSimpleDataSource;
import org.postgresql.jdbc.PgConnection;

import java.sql.SQLException;
import java.sql.Statement;

/**
 * Поток-получатель сообщений.
 */
public class MessageListener extends Thread {
    /**
     * DataSource для установки соединения с СУБД PostgreSQL.
     */
    private final PGSimpleDataSource dataSource;
    /**
     * Имя пользователя.
     */
    private final String username;

    public MessageListener(PGSimpleDataSource dataSource, String username) {
        this.dataSource = dataSource;
        this.username = username;
    }

    @Override
    public void run() {
        // 1. Устанавливаем соединение
        try(PgConnection connection = (PgConnection) dataSource.getConnection()) {
            // 2. Начало приема сообщений
            try (Statement statement = connection.createStatement()) {
                statement.execute("LISTEN " + username);
            }

            // 3. Запуск цикл чтения сообщений
            while (true) {
                // 4. Выполняем простейший запрос к СУБД, чтобы забрать оттуда последние сообщения
                try (Statement statement = connection.createStatement()) {
                    statement.execute("SELECT 1");
                }

                // 5. Читаем все новые сообщения и распечатываем их на консоль
                PGNotification[] notifications = connection.getNotifications();
                if (notifications != null) {
                    for (PGNotification notification : notifications) {
                        System.out.println("MESSAGE : " + notification.getParameter());
                    }
                }
                
                // 6. Немного подождем, прежде чем начать все заново
                Thread.sleep(100);
            }
        } catch (SQLException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}

Теперь приступим к потоку-отправителю. Поток отправитель должен ждать ввода текста с консоли, извлекать из него имя получателя и текст сообщения и отправлять его.
Для простоты будем считать, что первое слово строки - это имя пользователя, а все, что идет после имени через пробел - текст сообщения.

Тогда код потока-отправителя выглядит примерно так:

package ru.toolkas.thebestmessenger;

import org.postgresql.ds.PGSimpleDataSource;
import org.postgresql.jdbc.PgConnection;

import java.sql.SQLException;
import java.sql.Statement;
import java.util.Scanner;

public class MessageSender extends Thread {
    /**
     * DataSource для установки соединения с СУБД PostgreSQL.
     */
    private final PGSimpleDataSource dataSource;

    public MessageSender(PGSimpleDataSource dataSource) {
        this.dataSource = dataSource;
    }

    @Override
    public void run() {
        // 1. Читаем построчно ввод пользователя с консоли
        try (Scanner scanner = new Scanner(System.in)) {
            while (scanner.hasNextLine()) {
                String line = scanner.nextLine();

                if (line != null) {
                    // 2. Извлекаем имя пользователя и сообщение из введенной строки
                    int index = line.indexOf(" ");
                    String name = line.substring(0, index);
                    String message = line.substring(index + 1);

                    // 3. Устанавливаем соединение с СУБД
                    try (PgConnection connection = (PgConnection) dataSource.getConnection()) {
                        try (Statement statement = connection.createStatement()) {
                            // 4. Отправляем пользователю сообщение
                            statement.execute("NOTIFY " + name + ", '" + message + "'");
                        }
                    } catch (SQLException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }
}

Ну и последним шагом необходимо создать класс запуска нашего мессенджера. Это класс должен спрашивать у пользователя его имя, инициализировать DataSource для PostgreSQL и запускать потоки получателя и отправителя.

package ru.toolkas.thebestmessenger;

import org.postgresql.ds.PGSimpleDataSource;

import java.util.Scanner;

public class TheBestMessenger {
    public static void main(String[] args) throws InterruptedException {
        // 1. Запрашиваем имя пользователя
        System.out.println("What is your name?");

        // 2. Читаем введенное имя с консоли
        String name = new Scanner(System.in).nextLine();

        // 3. Приветствуем пользователя
        System.out.println("Hello, " + name);

        // 4. Инициализируем DataSource для соединения с СУБД PostgreSQL
        PGSimpleDataSource dataSource = new PGSimpleDataSource();
        // 4.1 Имя или IP сервера СУБД
        dataSource.setServerName("xxx");
        // 4.2 Порт сервера СУБД
        dataSource.setPortNumber(5432);
        // 4.3 Имя базы данных
        dataSource.setDatabaseName("postgres");
        // 4.4 Имя пользователя для установки соединения
        dataSource.setUser("postgres");
        // 4.5 Пароль пользователя для установки соединения
        dataSource.setPassword("123");

        // 5. Создаем поток-получатель
        MessageReceiver receiver = new MessageReceiver(dataSource, name);
        // 6. Создаем поток-отправитель
        MessageSender sender = new MessageSender(dataSource);

        // 7. Запускаем поток-получатель
        receiver.start();
        // 8. Запускаем поток-отправитель
        sender.start();

        // 9. Чтобы основная программа не завершилась раньше потоков - ждем завершения потока-отправителя
        sender.join();
    }
}

Теперь можно запустить программу и пообщаться с пользователем xxx. Диалог со стороны toolkas будет выглядеть так:

What is your name?
toolkas
Hello, toolkas
xxx Привет, xxx!
MESSAGE: И тебе привет!

А со стороны xxx так:

What is your name?
xxx
Hello, xxxx
MESSAGE: Привет, xxx!
toolkas И тебе привет!

Заключение

Итак, мы рассмотрели систему нотификации в PostgreSQL. Конечно, приведенный выше пример нельзя считать завершенным - этот мессенджер даже остановить нельзя :)

Но, надеюсь, он поможет вам решить вам сложные и важные задачи.

До новых встреч!