基于BottledWater-PG+nodejs实时地图应用实践-阿里云开发者社区

开发者社区> 遥想公瑾> 正文

基于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实现异步消息发送,该方法借助了表的触发器实现。应用程序是直连数据库且数据增删改都会走触发器。
  匆忙中,作者并未对比两者之间孰优孰劣,但一个直连库,一个间接消费,在不同需求中可选择一个比较符合要求的方案而加以应用。

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
基础设施的完善,推动应用形态不断变迁
我们把过去四十年分为五个重要的技术发展阶段,从时间轴上我们把它切分为:1980 - 2000,2000 - 2005,2005 - 2010,2010 - 2020 以及 2020 - 2025。物联网属于第五个阶段,从过去看未来,所以我们先回顾下整个技术发展史,先来看下前四个阶段分别经历了怎样的技术发展?在技术应用上的主要场景是什么?主流的应用形态是什么?诞生了什么样的新技术和新产品?在 198
56 0
【1026 - 1030直播导视 | PPT 下载】阿里专家直播:EasyTransfer平台应用实践、大数据+AI meetup 2020第二季·上海场来啦!
本周为大家带来了8场技术公开课,技术学习实践两不误!开源日迎来第六期啦线上对话阿里云专家:为大家带来EasyTransfer平台应用实践;大数据+AI meetup上海专场来啦!大家快办好小板凳坐等下周技术专场吧!
3454 0
函数计算助力高德地图平稳支撑亿级流量高峰
2020 年的“十一出行节”期间,高德地图创造了记录 ——截止 2020 年 10 月 1 日 13 时 27 分 27 秒,高德地图当日活跃用户突破 1 亿,比 2019 年 10 月 1 日提前 3 时 41 分达成此记录。 期间,Serverless 作为其中一个核心技术场景,平稳扛住了流量高峰期的考验。值得一提的是,由 Serverless 支撑的业务在流量高峰期的表现十分优秀,每分钟函数调用量接近两百万次。这再次验证了 Serverless 基础技术的价值,进一步拓展了技术场景。
3298 0
新一代图片编解码技术在淘宝的应用及落地
本文回顾淘宝图片发展的历史,阐述了新一代图像编解码格式AVIF在淘宝业务场景中的应用及落地方案,节省流量,为用户提供更好的看图体验。
64 0
+关注
遥想公瑾
个人专业是地理信息系统,拥有较丰富的GIS开发经验,对GIS在前端可视化和空间数据库(PostGIS,PostgreSQL)领域比较感兴趣。工作中致力于解决大量地理数据渲染问题,大量地理数据存储和空间查询优化。
39
文章
0
问答
文章排行榜
最热
最新
相关电子书
更多
《2021云上架构与运维峰会演讲合集》
立即下载
《零基础CSS入门教程》
立即下载
《零基础HTML入门教程》
立即下载