nanomsg:https://github.com/nanomsg/nanomsg

PipelineDB:https://github.com/pipelinedb/pipelinedb

nanomsgToPipelineDB:https://github.com/sangli00/nanomsgtopdb

創建extension

pipeline=# create extension nanomsgtopdb ;
CREATE EXTENSION

默認數據接收流

pipeline=# \d generic_stream
         Stream "public.generic_stream"
      
 
Column       |            Type
-------------------+-----------------------------
 data              | text
 arrival_timestamp | timestamp(0) with time zone

pipeline=#

創建Continuous View

pipeline=# create continuous view cv as select data from generic_stream;
CREATE CONTINUOUS VIEW
pipeline=# \d cv
Continuous 
 
view "public.cv"
 Column | Type | Modifiers
--------+------+-----------
 data   | text |

pipeline=#

發送數據:

[[email protected] test]# ./send_msg
connect to tcp://127.0.0.1:9999
Send msg size is 64

pipeline=# select * from cv;
 data
------
(0 rows)

pipeline=# select * from cv;
           data
 
--------------------------
 Hello ,from nanomsg msg.
(1 row)

pipeline=#

運用PostgreSQL來創建後臺進程,如圖所示：

多nanomsg worker process進行數據寫入，提高數據接收速度。

運用場景：

在底層數據采集的時候，原始數據直接寫入到Stream生成CV結果。

nanomsg 如何寫數據到PipelineDB