基于Postgres逻辑复制的推送式发件箱模式

CREATE TABLE publications(
-- the autoincremented position of the message to respect the order
position BIGSERIAL PRIMARY KEY,
-- this may allow you to partition publications, e.g. per tenant
publication_id VARCHAR(250) NOT NULL,
-- unique message id, which can be used for deduplication or idempotency
message_id VARCHAR(250) NOT NULL,
-- the message type, e.g. `TransactionRecorded`
message_type VARCHAR(250) NOT NULL,
-- serialised message data, e.g. to JSON
data JSONB NOT NULL,
-- diagnostic information of when the message was scheduled
scheduled TIMESTAMP WITH TIME ZONE NOT NULL default (now()),
);

CREATE TABLE subscribers
(
-- subscription name
subscription_id VARCHAR(250) PRIMARY KEY,
-- information about the position of the last processed message
last_processed_position INTEGER NULL,
-- inform which publication we're subscribing to
publication_id VARCHAR(250) NOT NULL,
);

SELECT
p.position, p.publication_id, p.message_id, p.message_type, [url=http://p.data/]p.data[/url], p.scheduled
FROM publication p
WHERE
p.publication_id = subscription_publication_id
p.last_processed_position > subscription_last_processed_position
LIMIT 100

wal_level = logical

version: "3"
services:
postgres:
image: postgres:14.5-alpine
ports:
- "5432:5432"
environment:
- POSTGRES_DB=postgres
- POSTGRES_PASSWORD=postgres
command:
- "postgres"
- "-c"
- "wal_level=logical"

CREATE PUBLICATION outbox_pub FOR TABLE outbox;

SELECT * FROM pg_create_logical_replication_slot('outbox_slot', 'pgoutput');

CREATE TABLE outbox (
position BIGSERIAL PRIMARY KEY,
event_type VARCHAR(250) NOT NULL,
data JSONB NOT NULL
);

public interface IEventsSubscription
{
IAsyncEnumerable<object> Subscribe(EventsSubscriptionOptions options, CancellationToken ct);
}
public record EventsSubscriptionOptions(
string ConnectionString,
string SlotName,
string PublicationName
);

var (connectionString, slotName, publicationName) = options;
await using var conn = new LogicalReplicationConnection(connectionString);
await [url=http://conn.open/]conn.Open[/url](ct);
var slot = new PgOutputReplicationSlot(slotName);

await foreach (var message in conn.StartReplication(slot, new PgOutputReplicationOptions(publicationName, 1), ct))
{
// TODO: PLACE PROCESSING LOGIC HERE
// Always call SetReplicationStatus() or assign LastAppliedLsn and LastFlushedLsn individually
// so that Npgsql can inform the server which WAL files can be removed/recycled.
conn.SetReplicationStatus(message.WalEnd);
// send the ACK to Postgres that we processed message
await conn.SendStatusUpdate(ct);
}

public class EventsSubscription: IEventsSubscription
{
    public async IAsyncEnumerable<object> Subscribe(EventsSubscriptionOptions options, [EnumeratorCancellation] CancellationToken ct)
    {
        var (connectionString, slotName, publicationName) = options;
        await using var conn = new LogicalReplicationConnection(connectionString);
        await conn.Open(ct);

        var slot = new PgOutputReplicationSlot(slotName);

        await foreach (var message in conn.StartReplication(slot, new PgOutputReplicationOptions(publicationName, 1), ct))
        {
            if (message is InsertMessage insertMessage)
            {
                yield return await InsertMessageHandler.Handle(insertMessage, ct);
            }

            conn.SetReplicationStatus(message.WalEnd);
            await conn.SendStatusUpdate(ct);
        }
    }
}

public static class InsertMessageHandler
{
    public static async Task<object> Handle(InsertMessage message, CancellationToken ct)
    {
        var columnNumber = 0;
        var eventTypeName = string.Empty;

        await foreach (var value in message.NewRow)
        {
            switch (columnNumber)
            {
                case 1:
                    eventTypeName = await value.GetTextReader().ReadToEndAsync();
                    break;
                case 2 when value.GetDataTypeName().ToLower() == "jsonb":
                {
                    var eventType = Type.GetType(eventTypeName);
                    if (eventType is null)
                        throw new ArgumentOutOfRangeException(nameof(eventType));

                    var @event = await JsonSerialization.FromJsonAsync(eventType, value.GetStream(), ct);

                    return @event!;
                }
            }

            columnNumber++;
        }

        throw new InvalidOperationException("You should not get here");
    }
}

var subscriptionOptions = new EventsSubscriptionOptions(ConnectrionString, "outbox_slot", "outbox_pub");
var subscription = new EventsSubscription();
await foreach (var readEvent in subscription.Subscribe(subscriptionOptions, ct))
{
// Handle upcoming event here, e.g.
// eventBus.Publish(readEvent);
}