基于BottledWater-PG+nodejs实时地图应用实践

简介: 前文作者讲述了BottledWater-PG安装部署,并在pg中实现了数据改变,向kafka发送消息的案例,详细参考《BottledWater-PG:PostgreSQL集成Kafka的实时数据交换平台》。

前文作者讲述了BottledWater-PG安装部署,并在pg中实现了数据改变,向kafka发送消息的案例,详细参考《BottledWater-PG:PostgreSQL集成Kafka的实时数据交换平台》。此前作者写过一篇pg的异步消息实现的实时地图应用案例《postgres+socket.io+nodejs实时地图应用实践》,本文将改用BottledWater-PG实现一遍。

一 服务器端

var fs = require('fs');
var http = require('http');
var socket = require('socket.io');
var Kafka = require('node-rdkafka');
    
var server = http.createServer(function(req, res) {
    
    res.writeHead(200, { 'Content-type': 'text/html'});
    res.end(fs.readFileSync(__dirname + '/index.html'));
    }).listen(8081, function() {
        console.log('Listening at: http://localhost:8081');
});
//注册socket.io
var socketio=socket.listen(server);
socketio.on('connection', function (socketclient) {
    console.log('已连接socket:');
    //socketclient.broadcast.emit('GPSCoor', data.payload);//广播给别人
    //socketclient.emit('GPSCoor', data.payload);//广播给自己
    
});

var consumer = new Kafka.KafkaConsumer({
  //'debug': 'all',
  'metadata.broker.list': '192.168.43.27:9092',
  'group.id': 'node-rdkafka-consumer-flow-example',
  'enable.auto.commit': false
});

var topicName = 'gps';

//logging debug messages, if debug is enabled 
consumer.on('event.log', function(log) {
  console.log(log);
});

//打印错误
consumer.on('error', function(err) {
  console.error('Error from consumer');
  console.error(err);
});

consumer.on('ready', function(arg) {
  console.log('consumer ready.' + JSON.stringify(arg));
  consumer.subscribe([topicName]);
  //准备消费消息
  consumer.consume();
});

consumer.on('data', function(m) {
    console.log(m);
    let _data;
    if(m.value==null)//delete操作发送来的消息
    {
        _data=JSON.parse(m.key);
        _data.tg_op='delete';
    }
    else{
        _data=m.value.toString();
        _data=JSON.parse(_data);
    }   
    console.log(_data);
    socketio.emit('GPSCoor', _data);//广播给所有的客户端
});

consumer.on('disconnected', function(arg) {
  console.log('consumer disconnected. ' + JSON.stringify(arg));
});

//启动
consumer.connect();

二 客户端

<html>
<head>
    <meta charset='utf-8'>
    <title>实时地图应用</title>
    <link rel="stylesheet" href="http://openlayers.org/en/v3.18.2/css/ol.css" type="text/css">
    <script src="http://openlayers.org/en/v3.18.2/build/ol.js"></script>
    <script src="/socket.io/socket.io.js"></script>
    <script>
            var wktform=new ol.format.WKT();//wkt解析
            var gpsSource=new ol.source.Vector();
            function init(){
                var gpsLayer=new ol.layer.Vector({
                    source:gpsSource,
                    style:new ol.style.Style({
                        image: new ol.style.Icon(({
                            anchor: [0.5, 1],
                            src: 'http://openlayers.org/en/v3.18.2/examples/data/icon.png'
                        }))
                    })
                });
                var map = new ol.Map({
                    layers : [
                        new ol.layer.Tile({
                            title : '街道图',
                            visible : true,
                            source : new ol.source.XYZ({
                                url : 'http://www.google.cn/maps/vt?pb=!1m5!1m4!1i{z}!2i{x}!3i{y}!4i256!2m3!1e0!2sm!3i342009817!3m9!2szh-CN!3sCN!5e18!12m1!1e47!12m3!1e37!2m1!1ssmartmaps!4e0&token=32965'
                            })
                        }),
                        gpsLayer
                    ],
                    target : 'map',
                    controls : ol.control.defaults({
                        attributionOptions : 
                        ({
                            collapsible : false
                        })
                    }),
                    view : new ol.View({
                        center : [0, 0],
                        zoom : 2
                    })
                });
                var iosocket = io.connect();
                //接受服务端消息
                iosocket.on('GPSCoor', function(data) {
                    console.log(data);
                    var id=data.id.int;
                    var feature;
                    if(data.tg_op=='delete'){
                        feature=gpsSource.getFeatureById(id);
                        if(feature)
                            gpsSource.removeFeature(feature);//删除点
                    }
                    else{
                        var geom=data.geom.string;
                        geom=wktform.readGeometry(geom);
                        geom.transform('EPSG:4326','EPSG:3857');
                        feature=gpsSource.getFeatureById(id);
                        if(feature)
                            feature.setGeometry(geom);//修改已有点
                        else{
                            feature=new ol.Feature({
                                geometry:geom
                            });
                            feature.setId(id);
                            gpsSource.addFeature(feature);//地图新增点
                        }
                    }
                });
            }
            
           
    </script>
</head>
<body onload="init()">
    <div id="map"></div>
</body>
</html>

三 测试成果

3.1 新增

mcsas=# insert into gps(name,geom) values ('opy','Point(118 31.5)');
INSERT 0 1
mcsas=# insert into gps(name,geom) values ('ty','Point(117 30.5)');
INSERT 0 1
img_fa5e408b18ab9db86a5d879c20f7dcc9.png
新增成果.png

3.2 修改

mcsas=# update gps set geom='Point(115 40)' where name='opy';
UPDATE 1
img_ab074c2cf274e1356c1df543d1e3acd1.png
修改成果.png

3.3 删除

mcsas=# delete from gps where name='opy';
DELETE 1

img_394f23ea6c9875f096f2a9012467a6bb.png
删除结果.png

四 总结

BottledWater-PG主要作用是将pg库中的表的增删改的消息都发往了kafka,应用程序并没有直接连接数据库,而是直接去消费kafka的消息。在表发生insert,update,delete能获取消息,但是truncate table并未向kafka生成消息,不知是否是我哪里遗漏。
  作者之前曾使用pg自带的notify与listen实现异步消息发送,该方法借助了表的触发器实现。应用程序是直连数据库且数据增删改都会走触发器。
  匆忙中,作者并未对比两者之间孰优孰劣,但一个直连库,一个间接消费,在不同需求中可选择一个比较符合要求的方案而加以应用。

相关文章
|
15天前
|
JavaScript NoSQL Redis
深入浅出:使用 Docker 容器化部署 Node.js 应用
在当今快速发展的软件开发领域,Docker 作为一种开源的容器化技术,已经成为了提高应用部署效率、实现环境一致性和便于维护的关键工具。本文将通过一个简单的 Node.js 应用示例,引导读者从零开始学习如何使用 Docker 容器化技术来部署应用。我们不仅会介绍 Docker 的基本概念和操作,还会探讨如何构建高效的 Docker 镜像,并通过 Docker Compose 管理多容器应用。此外,文章还将涉及到一些最佳实践,帮助读者更好地理解和应用 Docker 在日常开发和部署中的强大功能。
17 0
|
17天前
|
JavaScript 前端开发 网络协议
轻松搭建远程Node.js服务端,让你的应用在公共网络中畅行无阻!
Node.js 是能够在服务器端运行 JavaScript 的开放源代码、跨平台运行环境。Node.js 由 OpenJS Foundation(原为 Node.js Foundation,已与 JS Foundation 合并)持有和维护,亦为 Linux 基金会的项目。Node.js 采用 Google 开发的 V8 运行代码,使用事件驱动、非阻塞和异步输入输出模型等技术来提高性能,可优化应用程序的传输量和规模。这些技术通常用于资料密集的即时应用程序。
|
17天前
|
运维 JavaScript 开发者
深入浅出:使用Docker容器化部署Node.js应用
在当今快速发展的软件开发领域,构建一套高效、可靠且易于扩展的开发环境成为了许多开发者和企业的首要任务。本文将探讨如何利用Docker这一强大的容器化技术,实现对Node.js应用的快速部署和管理。不同于传统的摘要方式,我们将通过一个实际操作的视角,逐步引导读者理解Docker的基本概念、容器与镜像的区别、以及如何构建自己的Node.js应用Docker镜像,最终实现应用的容器化部署。此外,文章还将简要介绍Docker Compose的使用,帮助读者管理包含多个服务的复杂应用。无论是刚接触Docker的新手,还是希望深化理解容器化技术的资深开发者,本文都将提供有价值的见解。
13 0
|
23天前
|
JavaScript 前端开发 数据安全/隐私保护
NodeJS 下构建 命令行工具(CLI) 与 交互式命令界面 的实践
NodeJS 下构建 命令行工具(CLI) 与 交互式命令界面 的实践
139 1
|
23天前
|
JavaScript 前端开发 数据安全/隐私保护
深入探讨Node.js:构建高性能服务器端应用
深入探讨Node.js:构建高性能服务器端应用
156 1
|
2月前
|
负载均衡 JavaScript 算法
Node.js 多进程的概念、原理、优势以及如何使用多进程来提高应用程序的性能和可伸缩性
Node.js 多进程的概念、原理、优势以及如何使用多进程来提高应用程序的性能和可伸缩性
27 1
|
4月前
|
JavaScript 前端开发 Python
Node.js在Python中的应用实例demo
Node.js在Python中的应用实例demo
|
4月前
|
JavaScript 前端开发 API
Node.js在Python中的应用实例解析
Node.js在Python中的应用实例解析
|
4月前
|
弹性计算 开发框架 JavaScript
SAP UI5 应用开发教程之五十五 - 如何将本地 SAP UI5 应用通过 Node.js Express 部署到公网上试读版
SAP UI5 应用开发教程之五十五 - 如何将本地 SAP UI5 应用通过 Node.js Express 部署到公网上试读版
63 0
|
4月前
|
JavaScript 安全 应用服务中间件
Node.js 应用访问 https 服务器时遇到的错误消息 unable to get local issuer certificate
Node.js 应用访问 https 服务器时遇到的错误消息 unable to get local issuer certificate
56 1