该项目是用于内部系统总线的消息队列， 非FIFO，基于 PostgreSQL 的物化队列实现

--Message queue implemented by PG --@Author Anthony Chen --@Copyright reserved --2016-12-30 CREATE TABLE mq_config ( table_name name not null unique, channel name primary key ); CREATE TABLE mq_base ( msg_id bigserial not null, type_id bigint not null default 0, sent boolean not null default false , sent_at timestamp with time zone not null default now(), payload jsonb not null, delivered_at timestamp with time zone ,CONSTRAINT pk_mq_base PRIMARY KEY (msg_id) ); --Use of this index SET "enable_seqscan = OFF" should be executed to avoid seq scan CREATE UNIQUE INDEX idx_mq_base_not_sent ON mq_base USING btree (msg_id,type_id) WHERE sent = false; CREATE OR REPLACE FUNCTION mq_trigger_notify() RETURNS TRIGGER LANGUAGE PLPGSQL AS $$ DECLARE t_channel name; BEGIN SELECT channel INTO t_channel FROM mq_config WHERE table_name = TG_RELNAME; EXECUTE 'NOTIFY ' || quote_ident(t_channel) || ', ' || quote_literal(NEW.msg_id); RETURN NEW; END; $$; CREATE OR REPLACE FUNCTION mq_create_queue (in_channel text) RETURNS mq_config LANGUAGE PLPGSQL VOLATILE SECURITY DEFINER AS $$ DECLARE out_val mq_config%ROWTYPE; t_table_name name; BEGIN t_table_name := 'mq_queue_' || in_channel; INSERT INTO mq_config (table_name, channel) VALUES (t_table_name, in_channel) returning * into out_val; EXECUTE 'CREATE TABLE ' || quote_ident(t_table_name) || '( like mq_base INCLUDING ALL )'; EXECUTE ' GRANT ALL ON TABLE '||quote_ident(t_table_name) || ' TO public'; EXECUTE 'CREATE TRIGGER tr_mq_notify AFTER INSERT ON ' || quote_ident(t_table_name) || ' FOR EACH ROW EXECUTE PROCEDURE mq_trigger_notify()'; RETURN out_val; END; $$; REVOKE EXECUTE ON FUNCTION mq_create_queue(text) FROM public; CREATE OR REPLACE FUNCTION mq_drop_queue(in_channel name) RETURNS bool LANGUAGE plpgsql VOLATILE SECURITY DEFINER AS $$ declare t_table_name name; BEGIN SELECT table_name INTO t_table_name FROM mq_config WHERE channel = in_channel; EXECUTE 'DROP TABLE ' || quote_ident(t_table_name) || ' CASCADE'; DELETE FROM mq_config WHERE channel = in_channel; RETURN FOUND; END; $$; REVOKE EXECUTE ON FUNCTION mq_drop_queue(in_channel name) FROM public; CREATE OR REPLACE FUNCTION mq_send (in_channel text, in_payload jsonb,in_type bigint DEFAULT 0 ) RETURNS mq_base LANGUAGE PLPGSQL VOLATILE AS $$ DECLARE channel_entry mq_config%ROWTYPE; out_val mq_base%ROWTYPE; BEGIN SELECT * INTO channel_entry FROM mq_config WHERE channel = in_channel; IF NOT FOUND THEN RAISE EXCEPTION 'Channel Not Found'; END IF; EXECUTE 'INSERT INTO ' || quote_ident(channel_entry.table_name) || ' (type_id, payload) VALUES ( ' || in_type ||',' || quote_literal(in_payload) || '::jsonb ) RETURNING msg_id,type_id,sent, sent_at,payload, delivered_at' INTO out_val ; RETURN out_val; END; $$; CREATE OR REPLACE FUNCTION mq_recv( in_channel name, --Channel Name in_num_msgs bigint DEFAULT 1, --Number of message to get in_type_id bigint DEFAULT NULL::bigint --Message type id ) RETURNS SETOF mq_base LANGUAGE PLPGSQL VOLATILE AS $$ DECLARE channel_entry mq_config%ROWTYPE; BEGIN SELECT * INTO channel_entry FROM mq_config WHERE channel = in_channel; IF in_type_id is NULL THEN RETURN QUERY EXECUTE $e$ UPDATE $e$ || quote_ident(channel_entry.table_name) || $e$ SET delivered_at = now(),sent=true WHERE msg_id IN (SELECT msg_id FROM $e$ || quote_ident(channel_entry.table_name) || $e$ WHERE sent=false LIMIT $e$ || in_num_msgs || $e$ FOR UPDATE SKIP LOCKED ) RETURNING msg_id, type_id ,sent ,sent_at, payload, delivered_at $e$; ELSE RETURN QUERY EXECUTE $e$ UPDATE $e$ || quote_ident(channel_entry.table_name) || $e$ SET delivered_at = now(),sent=true WHERE msg_id IN (SELECT msg_id FROM $e$ || quote_ident(channel_entry.table_name) || $e$ WHERE sent=false and type_id= $e$ || in_type_id || $e$ LIMIT $e$ || in_num_msgs || $e$ FOR UPDATE SKIP LOCKED ) RETURNING msg_id, type_id ,sent ,sent_at, payload, delivered_at $e$; END IF; END; $$; COMMENT ON FUNCTION mq_send(text, jsonb, bigint) IS 'SET enable_seqscan = OFF; --TO BOOST PERFORMANCE--'; CREATE OR REPLACE FUNCTION mq_recv( in_channel name, --Channel Name in_type_rs text, --Message type id in_num_msgs bigint DEFAULT 1 --Number of message to get ) RETURNS SETOF mq_base LANGUAGE PLPGSQL VOLATILE AS $$ DECLARE channel_entry mq_config%ROWTYPE; BEGIN SELECT * INTO channel_entry FROM mq_config WHERE channel = in_channel; RETURN QUERY EXECUTE $e$ UPDATE $e$ || quote_ident(channel_entry.table_name) || $e$ SET delivered_at = now(),sent=true WHERE msg_id IN (SELECT msg_id FROM $e$ || quote_ident(channel_entry.table_name) || $e$ WHERE sent=false and type_id in $e$ || in_type_rs || $e$ LIMIT $e$ || in_num_msgs || $e$ FOR UPDATE SKIP LOCKED ) RETURNING msg_id, type_id ,sent ,sent_at, payload, delivered_at $e$; END; $$;