HBuilderX

HBuilderX

极客开发工具
uni-app

uni-app

开发一次,多端覆盖
uniCloud

uniCloud

云开发平台
HTML5+

HTML5+

增强HTML5的功能体验
MUI

MUI

上万Star的前端框架

★★★★★寻找iOS马甲包技术大大长期合作★★★★★

APP马甲包上架的技术熟练工们,请听我说帮我做马甲包,做一个ios的包,摩托变汽车,帮我做马甲包上架,瞬间变高富帅,屌丝也能崛起。骗子请让一让。需要大量的熟练技术人才,有的话请和我联系。拿出你的技术就会有回报。?
所有的技术们请联系我 QQ 51286559

继续阅读 »

APP马甲包上架的技术熟练工们,请听我说帮我做马甲包,做一个ios的包,摩托变汽车,帮我做马甲包上架,瞬间变高富帅,屌丝也能崛起。骗子请让一让。需要大量的熟练技术人才,有的话请和我联系。拿出你的技术就会有回报。?
所有的技术们请联系我 QQ 51286559

收起阅读 »

透传消息

如何设置透传消息样式跟推送通知一样

如何设置透传消息样式跟推送通知一样

微擎 、 人人商城 打包原生App交流、【登录、支付、分享...】

微信支付 微信登录

人人商城 打包原生App交流、【登录、支付、分享...】

懂的大佬一起分享交流。给后生们带带路.

人人商城 打包原生App交流、【登录、支付、分享...】

懂的大佬一起分享交流。给后生们带带路.

miniprogram-to-uniapp使用指南(各种小程序项目转换为uni-app项目)

qq小程序 支付宝小程序 百度小程序 钉钉小程序 微信小程序 小程序转成uni_app 小程序转为uni_app 小程序转uni_app 小程序转换

工具现在支持npm全局库、HBuilderX插件两种方式使用,任君选择,HBuilderX插件地址:https://ext.dcloud.net.cn/plugin?id=2656

一、它是谁?

【miniprogram-to-uniapp】转换微信小程序”项目为uni-app项目(新版本工具已经支持各种小程序转换)。

二、它的原理是什么?

最初是学了半节课堆和栈,觉得词法分析挺有意思的,再加上转换小程序插件时,发现这些繁琐的操作完全可以使用程序来完成。
核心是使用Babel获取AST(词法分析),然后或使用Babel自带函数增删,或正则分析替换等等操作。
可能有的朋友觉得,这种为啥不是纯正则分析? 纯正则可以解决一部分问题,对于标签与标签内容,其实正则是很难进行区分的哈。

三、它能做哪些事情?

  • 支持微信、QQ、头条/抖音、支付宝/钉钉和百度等小程序转换到 uni-app 项目
  • 支持有/无云开发的小程序项目转换为 uni-app 项目(cloudfunctions 目录将被忽略,uni-app 结合小程序云开发见:使用 uni-app 进行微信小程序云开发经验分享)
  • 支持解析 TypeScript 小程序项目
  • 支持解析使用 npm 模块的小程序项目
  • 支持解析 include 标签
  • 支持解析 template 标签
  • 支持解析 Behavior 文件为 mixins 文件
  • 支持.js', .wxml 和*.wxss 文件进行相应转换,并做了大量的优化
  • 支持识别 App、Page、Component、VantComponent、Behavior 和纯 Javascript 文件的转换
  • 修复变量名与函数重名的情况
  • 合并使用 require 导入的 wxs 文件
  • setData() polyfill
  • 搜索未在 data 声明,而直接在 setData()里使用的变量,并修复
  • 使用jyf-parser替换 wxParse(感谢网友 “爱瑞巴勒康忙北鼻” 的建议)
  • 因 uni-app 会将所有非 static 目录的资源文件删除,因此将所有资源文件移入 static 目录,并修复所有能修复到的路径(目前 uni 编译时会将非 static 目录的文件复制一份到 static 目录,但并不完全,因此本功能仍保留)

四、它还有哪些不支持转换?

  • 不支持转换反编译后的小程序项目
  • 不支持转换使用 uni-app 编译的小程序项目
  • 不支持转换使用 redux 开发的小程序(代表为:网易云信小程序 DEMO)
  • 不支持转换使用 wxpage 开发的小程序(https://github.com/tvfe/wxpage)
  • 不支持转换使用腾讯 omi 开发的小程序(https://github.com/Tencent/omi)
  • 不支持转换小程序抽象节点 componentGenerics
  • 不支持 component 里的 pageLifetimes 生命周期,请手动绕过
  • 不支持使用 js 系统关键字作为函数或变量名(如 default、import、return、switch 等)
  • 不支持以\$开头的变量名称,如 Page({data:{$data:{name:"hello"}}}) ,刚好\$data 是 vue 内置变量,so 不支持,需手动修复
  • 不支持以动态绑定的函数<input @input="test{{index+1}}">,需手动修复
  • 更多,请参照miniprogram to uniapp 工具答疑

五、怎么使用?

第一步

在命令行里,运行【 npm install miniprogram-to-uniapp -g 】进行安装,因为这个包是工具,要求全局都能使用,所以需要-g进行全局安装。
如果运行npm报错,请先安装Node.js,下载地址:https://nodejs.org/zh-cn/

第二步

继续在命令行里,运行【 wtu -V 】,执行结果如下:

显示版本号,说明已经安装成功了。(wtu -> 取自wx to uni之意,后面都用这个全局命令)

第三步

在命令行里,输入【wtu -i "你的小程序项目路径"】
注意 -i 前面和后面都有空格!!!
注意 -i 前面和后面都有空格!!!
注意 -i 前面和后面都有空格!!!
如:【wtu -i "E:\zpWork\Project_self\miniprogram-to-uniapp\test\test-wx-to-uni"】 ,回车后即可以在源项目同及目录得到一个后缀为_uni的目录,即转换成功。

转换前:

转换后:

转换后的项目文件对比(左边是小程序项目,右边是Uni-app项目目录):

第四步

将转换后的xxx_uni项目导入到hbuilder X,
点击菜单 运行--> 运行到小程序模拟器-->微信开发者工具!(如果是使用了vant的项目,请运行到H5,vant项目转换后仅支持H5和app)
点击菜单 运行--> 运行到小程序模拟器-->微信开发者工具!(如果是使用了vant的项目,请运行到H5,vant项目转换后仅支持H5和app)
点击菜单 运行--> 运行到小程序模拟器-->微信开发者工具!(如果是使用了vant的项目,请运行到H5,vant项目转换后仅支持H5和app)
重要的话说三遍!
然后查看转换后的项目运行到小程序,是否可以正常运行无报错!
(因为这种转换非100%,所以至少需要保证 小程序-->uniapp-->小程序仍然能正常运行,再考虑运行到其他小程序或app),
如有报错,请根据miniprogram to uniapp 工具答疑 进行修改,保证无报错,然后再运行到其他平台。

工具升级

因为工具更新比较频繁,安装后,可以使用如下命令进行升级:
npm update miniprogram-to-uniapp -g

注意:

  1. 遇到问题请参考文档:miniprogram to uniapp 工具答疑
  2. github上面的只是源代码。 日常使用,"仅"需要运行【 npm install miniprogram-to-uniapp -g 】进行安装才行,无需下载工具源码。
  3. 每次转换时都会将输出目录清空,请注意代码保存。
  4. 后续会有版本更新,请及时关注并更新~~,可以关注QQ群:780359397 获取最新信息

对于使用有疑问或建议,可以加入QQ群:780359397 进行讨论。

<a target="_blank" href="http://shang.qq.com/wpa/qunwpa?idkey=6cccd111e447ed70ee0c17672a452bf71e7e62cfa6b427bbd746df2d32297b64"><img border="0" src="http://pub.idqqimg.com/wpa/images/group.png" alt="小程序转uni-app讨论群" title="小程序转uni-app讨论群"></a>

因为有朋友使用本工具导致自己文件被删,在此表示抱歉!

this.setData()代码出处:https://ask.dcloud.net.cn/article/35020,在些表示感谢~

转换工具源码:https://github.com/zhangdaren/miniprogram-to-uniapp ,欢迎star和建议~

如有遇到bug或问题,欢迎评论、给出改进建议,感谢~~

继续阅读 »

工具现在支持npm全局库、HBuilderX插件两种方式使用,任君选择,HBuilderX插件地址:https://ext.dcloud.net.cn/plugin?id=2656

一、它是谁?

【miniprogram-to-uniapp】转换微信小程序”项目为uni-app项目(新版本工具已经支持各种小程序转换)。

二、它的原理是什么?

最初是学了半节课堆和栈,觉得词法分析挺有意思的,再加上转换小程序插件时,发现这些繁琐的操作完全可以使用程序来完成。
核心是使用Babel获取AST(词法分析),然后或使用Babel自带函数增删,或正则分析替换等等操作。
可能有的朋友觉得,这种为啥不是纯正则分析? 纯正则可以解决一部分问题,对于标签与标签内容,其实正则是很难进行区分的哈。

三、它能做哪些事情?

  • 支持微信、QQ、头条/抖音、支付宝/钉钉和百度等小程序转换到 uni-app 项目
  • 支持有/无云开发的小程序项目转换为 uni-app 项目(cloudfunctions 目录将被忽略,uni-app 结合小程序云开发见:使用 uni-app 进行微信小程序云开发经验分享)
  • 支持解析 TypeScript 小程序项目
  • 支持解析使用 npm 模块的小程序项目
  • 支持解析 include 标签
  • 支持解析 template 标签
  • 支持解析 Behavior 文件为 mixins 文件
  • 支持.js', .wxml 和*.wxss 文件进行相应转换,并做了大量的优化
  • 支持识别 App、Page、Component、VantComponent、Behavior 和纯 Javascript 文件的转换
  • 修复变量名与函数重名的情况
  • 合并使用 require 导入的 wxs 文件
  • setData() polyfill
  • 搜索未在 data 声明,而直接在 setData()里使用的变量,并修复
  • 使用jyf-parser替换 wxParse(感谢网友 “爱瑞巴勒康忙北鼻” 的建议)
  • 因 uni-app 会将所有非 static 目录的资源文件删除,因此将所有资源文件移入 static 目录,并修复所有能修复到的路径(目前 uni 编译时会将非 static 目录的文件复制一份到 static 目录,但并不完全,因此本功能仍保留)

四、它还有哪些不支持转换?

  • 不支持转换反编译后的小程序项目
  • 不支持转换使用 uni-app 编译的小程序项目
  • 不支持转换使用 redux 开发的小程序(代表为:网易云信小程序 DEMO)
  • 不支持转换使用 wxpage 开发的小程序(https://github.com/tvfe/wxpage)
  • 不支持转换使用腾讯 omi 开发的小程序(https://github.com/Tencent/omi)
  • 不支持转换小程序抽象节点 componentGenerics
  • 不支持 component 里的 pageLifetimes 生命周期,请手动绕过
  • 不支持使用 js 系统关键字作为函数或变量名(如 default、import、return、switch 等)
  • 不支持以\$开头的变量名称,如 Page({data:{$data:{name:"hello"}}}) ,刚好\$data 是 vue 内置变量,so 不支持,需手动修复
  • 不支持以动态绑定的函数<input @input="test{{index+1}}">,需手动修复
  • 更多,请参照miniprogram to uniapp 工具答疑

五、怎么使用?

第一步

在命令行里,运行【 npm install miniprogram-to-uniapp -g 】进行安装,因为这个包是工具,要求全局都能使用,所以需要-g进行全局安装。
如果运行npm报错,请先安装Node.js,下载地址:https://nodejs.org/zh-cn/

第二步

继续在命令行里,运行【 wtu -V 】,执行结果如下:

显示版本号,说明已经安装成功了。(wtu -> 取自wx to uni之意,后面都用这个全局命令)

第三步

在命令行里,输入【wtu -i "你的小程序项目路径"】
注意 -i 前面和后面都有空格!!!
注意 -i 前面和后面都有空格!!!
注意 -i 前面和后面都有空格!!!
如:【wtu -i "E:\zpWork\Project_self\miniprogram-to-uniapp\test\test-wx-to-uni"】 ,回车后即可以在源项目同及目录得到一个后缀为_uni的目录,即转换成功。

转换前:

转换后:

转换后的项目文件对比(左边是小程序项目,右边是Uni-app项目目录):

第四步

将转换后的xxx_uni项目导入到hbuilder X,
点击菜单 运行--> 运行到小程序模拟器-->微信开发者工具!(如果是使用了vant的项目,请运行到H5,vant项目转换后仅支持H5和app)
点击菜单 运行--> 运行到小程序模拟器-->微信开发者工具!(如果是使用了vant的项目,请运行到H5,vant项目转换后仅支持H5和app)
点击菜单 运行--> 运行到小程序模拟器-->微信开发者工具!(如果是使用了vant的项目,请运行到H5,vant项目转换后仅支持H5和app)
重要的话说三遍!
然后查看转换后的项目运行到小程序,是否可以正常运行无报错!
(因为这种转换非100%,所以至少需要保证 小程序-->uniapp-->小程序仍然能正常运行,再考虑运行到其他小程序或app),
如有报错,请根据miniprogram to uniapp 工具答疑 进行修改,保证无报错,然后再运行到其他平台。

工具升级

因为工具更新比较频繁,安装后,可以使用如下命令进行升级:
npm update miniprogram-to-uniapp -g

注意:

  1. 遇到问题请参考文档:miniprogram to uniapp 工具答疑
  2. github上面的只是源代码。 日常使用,"仅"需要运行【 npm install miniprogram-to-uniapp -g 】进行安装才行,无需下载工具源码。
  3. 每次转换时都会将输出目录清空,请注意代码保存。
  4. 后续会有版本更新,请及时关注并更新~~,可以关注QQ群:780359397 获取最新信息

对于使用有疑问或建议,可以加入QQ群:780359397 进行讨论。

<a target="_blank" href="http://shang.qq.com/wpa/qunwpa?idkey=6cccd111e447ed70ee0c17672a452bf71e7e62cfa6b427bbd746df2d32297b64"><img border="0" src="http://pub.idqqimg.com/wpa/images/group.png" alt="小程序转uni-app讨论群" title="小程序转uni-app讨论群"></a>

因为有朋友使用本工具导致自己文件被删,在此表示抱歉!

this.setData()代码出处:https://ask.dcloud.net.cn/article/35020,在些表示感谢~

转换工具源码:https://github.com/zhangdaren/miniprogram-to-uniapp ,欢迎star和建议~

如有遇到bug或问题,欢迎评论、给出改进建议,感谢~~

收起阅读 »

uni-app开发问题总结(小程序与h5)1:tabbar与redirectTo的混合使用

https://blog.csdn.net/tan9374/article/details/92990494

https://blog.csdn.net/tan9374/article/details/92990494

2019.6.20(改进版本)mui通过native.js调用安卓原生类库,通过广播方式实现扫码结果的接收

h5+ java DCloud mui
<script type="text/javascript" src="js/common.js"> </script>  
    <script>  
        mui.plusReady(function() {  
            var receiver;  
            main = plus.android.runtimeMainActivity(); //获取activity    
            receiver = plus.android.implements('io.dcloud.android.content.BroadcastReceiver', {  
                onReceive: function(context, intent) { //实现onReceiver回调函数    
                // alert(1);  
                plus.android.importClass(intent); //通过intent实例引入intent类,方便以后的‘.’操作    
                alert(intent.getStringExtra("scannerdata")); //获取数据  
                // main.unregisterReceiver(receiver); //取消监听    
                }  
            });  
            var IntentFilter = plus.android.importClass('android.content.IntentFilter');  
            var Intent = plus.android.importClass('android.content.Intent');  
            var filter = new IntentFilter();  
            filter.addAction("com.android.server.scannerservice.broadcast"); //监听扫码广播    
            main.registerReceiver(receiver, filter); //注册监听  
            console.log('ok');  
        });  
    </script>  

改自 http://ask.dcloud.net.cn/article/12822

//监听扫码广播
//获取数据
以上数据测试用东大集成pda
mui 接收系统广播
有问题联系本人qq 659760615

common.js下载地址 点个star 谢谢啦
https://gitee.com/czxhehe/broadcasting/tree/master/

继续阅读 »
<script type="text/javascript" src="js/common.js"> </script>  
    <script>  
        mui.plusReady(function() {  
            var receiver;  
            main = plus.android.runtimeMainActivity(); //获取activity    
            receiver = plus.android.implements('io.dcloud.android.content.BroadcastReceiver', {  
                onReceive: function(context, intent) { //实现onReceiver回调函数    
                // alert(1);  
                plus.android.importClass(intent); //通过intent实例引入intent类,方便以后的‘.’操作    
                alert(intent.getStringExtra("scannerdata")); //获取数据  
                // main.unregisterReceiver(receiver); //取消监听    
                }  
            });  
            var IntentFilter = plus.android.importClass('android.content.IntentFilter');  
            var Intent = plus.android.importClass('android.content.Intent');  
            var filter = new IntentFilter();  
            filter.addAction("com.android.server.scannerservice.broadcast"); //监听扫码广播    
            main.registerReceiver(receiver, filter); //注册监听  
            console.log('ok');  
        });  
    </script>  

改自 http://ask.dcloud.net.cn/article/12822

//监听扫码广播
//获取数据
以上数据测试用东大集成pda
mui 接收系统广播
有问题联系本人qq 659760615

common.js下载地址 点个star 谢谢啦
https://gitee.com/czxhehe/broadcasting/tree/master/

收起阅读 »

uni-app / mui / h5+ 手机底部导航栏(虚拟按键菜单)背景颜色

onLaunch() {

                   //底部导航栏(虚拟按键)背景颜色  
        var color = plus.android.newObject("android.graphics.Color")  
        var ac = plus.android.runtimeMainActivity();  
        var c2int = plus.android.invoke(color, "parseColor", "#161516");  
        console.log("c2int===" + JSON.stringify(c2int))  
        var win = plus.android.invoke(ac, "getWindow");  
        console.log("win===" + JSON.stringify(win))  
        plus.android.invoke(win, "setNavigationBarColor", c2int);  

继续阅读 »

onLaunch() {

                   //底部导航栏(虚拟按键)背景颜色  
        var color = plus.android.newObject("android.graphics.Color")  
        var ac = plus.android.runtimeMainActivity();  
        var c2int = plus.android.invoke(color, "parseColor", "#161516");  
        console.log("c2int===" + JSON.stringify(c2int))  
        var win = plus.android.invoke(ac, "getWindow");  
        console.log("win===" + JSON.stringify(win))  
        plus.android.invoke(win, "setNavigationBarColor", c2int);  

收起阅读 »

如何判断时候安装了微信、qq等

技术分享

案例判断是否安装微信 其他同理
if (plus.runtime.isApplicationExist({ pname: 'com.tencent.mm', action: 'weixin://' })) {
//安装了微信
} else {
//未安装微信
}

继续阅读 »

案例判断是否安装微信 其他同理
if (plus.runtime.isApplicationExist({ pname: 'com.tencent.mm', action: 'weixin://' })) {
//安装了微信
} else {
//未安装微信
}

收起阅读 »

Spark Streaming的优化之路——从Receiver到Direct模式

数据交互


作者:个推数据研发工程师 学长

1 业务背景

随着大数据的快速发展,业务场景越来越复杂,离线式的批处理框架MapReduce已经不能满足业务,大量的场景需要实时的数据处理结果来进行分析、决策。Spark Streaming是一种分布式的大数据实时计算框架,他提供了动态的,高吞吐量的,可容错的流式数据处理,不仅可以实现用户行为分析,还能在金融、舆情分析、网络监控等方面发挥作用。个推开发者服务——消息推送“应景推送”正是应用了Spark Streaming技术,基于大数据分析人群属性,同时利用LBS地理围栏技术,实时触发精准消息推送,实现用户的精细化运营。此外,个推在应用Spark Streaming做实时处理kafka数据时,采用Direct模式代替Receiver模式的手段,实现了资源优化和程序稳定性提升。

本文将从Spark Streaming获取kafka数据的两种模式入手,结合个推实践,带你解读Receiver和Direct模式的原理和特点,以及从Receiver模式到Direct模式的优化对比。

2 两种模式的原理和区别

Receiver模式

1. Receiver模式下的运行架构


1)InputDStream: 从流数据源接收的输入数据。
2)Receiver:负责接收数据流,并将数据写到本地。
3)Streaming Context:代表SparkStreaming,负责Streaming层面的任务调度,生成jobs发送到Spark engine处理。
4)Spark Context: 代表Spark Core,负责批处理层面的任务调度,真正执行job的Spark engine。

2. Receiver从kafka拉取数据的过程


该模式下:
1)在executor上会有receiver从kafka接收数据并存储在Spark executor中,在到了batch时间后触发job去处理接收到的数据,1个receiver占用1个core;
2)为了不丢数据需要开启WAL机制,这会将receiver接收到的数据写一份备份到第三方系统上(如:HDFS);
3)receiver内部使用kafka High Level API去消费数据及自动更新offset。

Direct模式

1. Direct模式下的运行架构

与receiver模式类似,不同在于executor中没有receiver组件,从kafka拉去数据的方式不同。

2. Direct从kafka拉取数据的过程


该模式下:
1)没有receiver,无需额外的core用于不停地接收数据,而是定期查询kafka中的每个partition的最新的offset,每个批次拉取上次处理的offset和当前查询的offset的范围的数据进行处理;
2)为了不丢数据,无需将数据备份落地,而只需要手动保存offset即可;
3)内部使用kafka simple Level API去消费数据, 需要手动维护offset,kafka zk上不会自动更新offset。

Receiver与Direct模式的区别

1.前者在executor中有Receiver接受数据,并且1个Receiver占用一个core;而后者无Receiver,所以不会暂用core;
2.前者InputDStream的分区是 num_receiver *batchInterval/blockInteral,后者的分区数是kafka topic partition的数量。Receiver模式下num_receiver的设置不合理会影响性能或造成资源浪费;如果设置太小,并行度不够,整个链路上接收数据将是瓶颈;如果设置太多,则会浪费资源;
3.前者使用zookeeper来维护consumer的偏移量,而后者需要自己维护偏移量;
4.为了保证不丢失数据,前者需要开启WAL机制,而后者不需要,只需要在程序中成功消费完数据后再更新偏移量即可。

3 Receiver改造成Direct模式

个推使用Spark Streaming做实时处理kafka数据,先前使用的是receiver模式;

receiver有以下特点
1.receiver模式下,每个receiver需要单独占用一个core;
2.为了保证不丢失数据,需要开启WAL机制,使用checkpoint保存状态;
3.当receiver接受数据速率大于处理数据速率,导致数据积压,最终可能会导致程序挂掉。

由于以上特点,receiver模式下会造成一定的资源浪费;使用checkpoint保存状态, 如果需要升级程序,则会导致checkpoint无法使用;第3点receiver模式下会导致程序不太稳定;并且如果设置receiver数量不合理也会造成性能瓶颈在receiver。为了优化资源和程序稳定性,应将receiver模式改造成direct模式。

修改方式如下:

1. 修改InputDStream的创建

将receiver的:

val kafkaStream = KafkaUtils.createStream(streamingContext,  
     [ZK quorum], [consumer group id], [per-topic number of Kafka partitions to consume])

改成direct的:

val directKafkaStream = KafkaUtils.createDirectStream[  
     [key class], [value class], [key decoder class], [value decoder class] ](  
     streamingContext, [map of Kafka parameters], [set of topics to consume])

2. 手动维护offset

receiver模式代码:
(receiver模式不需要手动维护offset,而是内部通过kafka consumer high level API 提交到kafka/zk保存)

kafkaStream.map {  
           ...  
 }.foreachRDD { rdd =>  
    // 数据处理  
    doCompute(rdd)  
 }

direct模式代码:

directKafkaStream.map {  
           ...  
 }.foreachRDD { rdd =>  
    // 获取当前rdd数据对应的offset  
    val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges  
    // 数据处理  
    doCompute(rdd)  
    // 自己实现保存offset  
    commitOffsets(offsetRanges)  
 }

4 其他优化点

1. 在receiver模式下
1)拆分InputDStream,增加Receiver,从而增加接收数据的并行度;
2)调整blockInterval,适当减小,增加task数量,从而增加并行度(在core的数量>task数量的情况下);
3)如果开启了WAL机制,数据的存储级别设置为MOMERY_AND_DISK_SER。

2.数据序列化使用Kryoserializationl,相比Java serializationl 更快,序列化后的数据更小;

3.建议使用CMS垃圾回收器降低GC开销;

4.选择高性能的算子(mapPartitions, foreachPartitions, aggregateByKey等);

5.repartition的使用:在streaming程序中因为batch时间特别短,所以数据量一般较小,所以repartition的时间短,可以解决一些因为topicpartition中数据分配不均匀导致的数据倾斜问题;

6.因为SparkStreaming生产的job最终都是在sparkcore上运行的,所以sparkCore的优化也很重要;

7.BackPressure流控

1)为什么引入Backpressure?
当batch processing time>batchinterval 这种情况持续过长的时间,会造成数据在内存中堆积,导致Receiver所在Executor内存溢出等问题;

2)Backpressure:根据JobScheduler反馈作业的执行信息来动态调整数据接收率;

3)配置使用:

spark.streaming.backpressure.enabled  
含义: 是否启用 SparkStreaming内部的backpressure机制,  
默认值:false ,表示禁用  

spark.streaming.backpressure.initialRate  
含义: receiver 为第一个batch接收数据时的比率  

spark.streaming.receiver.maxRate  
含义: receiver接收数据的最大比率,如果设置值<=0, 则receiver接收数据比率不受限制  

spark.streaming.kafka.maxRatePerPartition  
含义: 从每个kafka partition中读取数据的最大比率

8.speculation机制
spark内置speculation机制,推测job中的运行特别慢的task,将这些task kill,并重新调度这些task执行。
默认speculation机制是关闭的,通过以下配置参数开启:

spark.speculation=true

注意:在有些情况下,开启speculation反而效果不好,比如:streaming程序消费多个topic时,从kafka读取数据直接处理,没有重新分区,这时如果多个topic的partition的数据量相差较大那么可能会导致正常执行更大数据量的task会被认为执行缓慢,而被中途kill掉,这种情况下可能导致batch的处理时间反而变长;可以通过repartition来解决这个问题,但是要衡量repartition的时间;而在streaming程序中因为batch时间特别短,所以数据量一般较小,所以repartition的时间短,不像spark_batch一次处理大量数据一旦repartition则会特别久,所以最终还是要根据具体情况测试来决定。

5 总结

将Receiver模式改成Direct模式,实现了资源优化,提升了程序的稳定性,缺点是需要自己管理offset,操作相对复杂。未来,个推将不断探索和优化Spark Streaming技术,发挥其强大的数据处理能力,为建设实时数仓提供保障。

继续阅读 »


作者:个推数据研发工程师 学长

1 业务背景

随着大数据的快速发展,业务场景越来越复杂,离线式的批处理框架MapReduce已经不能满足业务,大量的场景需要实时的数据处理结果来进行分析、决策。Spark Streaming是一种分布式的大数据实时计算框架,他提供了动态的,高吞吐量的,可容错的流式数据处理,不仅可以实现用户行为分析,还能在金融、舆情分析、网络监控等方面发挥作用。个推开发者服务——消息推送“应景推送”正是应用了Spark Streaming技术,基于大数据分析人群属性,同时利用LBS地理围栏技术,实时触发精准消息推送,实现用户的精细化运营。此外,个推在应用Spark Streaming做实时处理kafka数据时,采用Direct模式代替Receiver模式的手段,实现了资源优化和程序稳定性提升。

本文将从Spark Streaming获取kafka数据的两种模式入手,结合个推实践,带你解读Receiver和Direct模式的原理和特点,以及从Receiver模式到Direct模式的优化对比。

2 两种模式的原理和区别

Receiver模式

1. Receiver模式下的运行架构


1)InputDStream: 从流数据源接收的输入数据。
2)Receiver:负责接收数据流,并将数据写到本地。
3)Streaming Context:代表SparkStreaming,负责Streaming层面的任务调度,生成jobs发送到Spark engine处理。
4)Spark Context: 代表Spark Core,负责批处理层面的任务调度,真正执行job的Spark engine。

2. Receiver从kafka拉取数据的过程


该模式下:
1)在executor上会有receiver从kafka接收数据并存储在Spark executor中,在到了batch时间后触发job去处理接收到的数据,1个receiver占用1个core;
2)为了不丢数据需要开启WAL机制,这会将receiver接收到的数据写一份备份到第三方系统上(如:HDFS);
3)receiver内部使用kafka High Level API去消费数据及自动更新offset。

Direct模式

1. Direct模式下的运行架构

与receiver模式类似,不同在于executor中没有receiver组件,从kafka拉去数据的方式不同。

2. Direct从kafka拉取数据的过程


该模式下:
1)没有receiver,无需额外的core用于不停地接收数据,而是定期查询kafka中的每个partition的最新的offset,每个批次拉取上次处理的offset和当前查询的offset的范围的数据进行处理;
2)为了不丢数据,无需将数据备份落地,而只需要手动保存offset即可;
3)内部使用kafka simple Level API去消费数据, 需要手动维护offset,kafka zk上不会自动更新offset。

Receiver与Direct模式的区别

1.前者在executor中有Receiver接受数据,并且1个Receiver占用一个core;而后者无Receiver,所以不会暂用core;
2.前者InputDStream的分区是 num_receiver *batchInterval/blockInteral,后者的分区数是kafka topic partition的数量。Receiver模式下num_receiver的设置不合理会影响性能或造成资源浪费;如果设置太小,并行度不够,整个链路上接收数据将是瓶颈;如果设置太多,则会浪费资源;
3.前者使用zookeeper来维护consumer的偏移量,而后者需要自己维护偏移量;
4.为了保证不丢失数据,前者需要开启WAL机制,而后者不需要,只需要在程序中成功消费完数据后再更新偏移量即可。

3 Receiver改造成Direct模式

个推使用Spark Streaming做实时处理kafka数据,先前使用的是receiver模式;

receiver有以下特点
1.receiver模式下,每个receiver需要单独占用一个core;
2.为了保证不丢失数据,需要开启WAL机制,使用checkpoint保存状态;
3.当receiver接受数据速率大于处理数据速率,导致数据积压,最终可能会导致程序挂掉。

由于以上特点,receiver模式下会造成一定的资源浪费;使用checkpoint保存状态, 如果需要升级程序,则会导致checkpoint无法使用;第3点receiver模式下会导致程序不太稳定;并且如果设置receiver数量不合理也会造成性能瓶颈在receiver。为了优化资源和程序稳定性,应将receiver模式改造成direct模式。

修改方式如下:

1. 修改InputDStream的创建

将receiver的:

val kafkaStream = KafkaUtils.createStream(streamingContext,  
     [ZK quorum], [consumer group id], [per-topic number of Kafka partitions to consume])

改成direct的:

val directKafkaStream = KafkaUtils.createDirectStream[  
     [key class], [value class], [key decoder class], [value decoder class] ](  
     streamingContext, [map of Kafka parameters], [set of topics to consume])

2. 手动维护offset

receiver模式代码:
(receiver模式不需要手动维护offset,而是内部通过kafka consumer high level API 提交到kafka/zk保存)

kafkaStream.map {  
           ...  
 }.foreachRDD { rdd =>  
    // 数据处理  
    doCompute(rdd)  
 }

direct模式代码:

directKafkaStream.map {  
           ...  
 }.foreachRDD { rdd =>  
    // 获取当前rdd数据对应的offset  
    val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges  
    // 数据处理  
    doCompute(rdd)  
    // 自己实现保存offset  
    commitOffsets(offsetRanges)  
 }

4 其他优化点

1. 在receiver模式下
1)拆分InputDStream,增加Receiver,从而增加接收数据的并行度;
2)调整blockInterval,适当减小,增加task数量,从而增加并行度(在core的数量>task数量的情况下);
3)如果开启了WAL机制,数据的存储级别设置为MOMERY_AND_DISK_SER。

2.数据序列化使用Kryoserializationl,相比Java serializationl 更快,序列化后的数据更小;

3.建议使用CMS垃圾回收器降低GC开销;

4.选择高性能的算子(mapPartitions, foreachPartitions, aggregateByKey等);

5.repartition的使用:在streaming程序中因为batch时间特别短,所以数据量一般较小,所以repartition的时间短,可以解决一些因为topicpartition中数据分配不均匀导致的数据倾斜问题;

6.因为SparkStreaming生产的job最终都是在sparkcore上运行的,所以sparkCore的优化也很重要;

7.BackPressure流控

1)为什么引入Backpressure?
当batch processing time>batchinterval 这种情况持续过长的时间,会造成数据在内存中堆积,导致Receiver所在Executor内存溢出等问题;

2)Backpressure:根据JobScheduler反馈作业的执行信息来动态调整数据接收率;

3)配置使用:

spark.streaming.backpressure.enabled  
含义: 是否启用 SparkStreaming内部的backpressure机制,  
默认值:false ,表示禁用  

spark.streaming.backpressure.initialRate  
含义: receiver 为第一个batch接收数据时的比率  

spark.streaming.receiver.maxRate  
含义: receiver接收数据的最大比率,如果设置值<=0, 则receiver接收数据比率不受限制  

spark.streaming.kafka.maxRatePerPartition  
含义: 从每个kafka partition中读取数据的最大比率

8.speculation机制
spark内置speculation机制,推测job中的运行特别慢的task,将这些task kill,并重新调度这些task执行。
默认speculation机制是关闭的,通过以下配置参数开启:

spark.speculation=true

注意:在有些情况下,开启speculation反而效果不好,比如:streaming程序消费多个topic时,从kafka读取数据直接处理,没有重新分区,这时如果多个topic的partition的数据量相差较大那么可能会导致正常执行更大数据量的task会被认为执行缓慢,而被中途kill掉,这种情况下可能导致batch的处理时间反而变长;可以通过repartition来解决这个问题,但是要衡量repartition的时间;而在streaming程序中因为batch时间特别短,所以数据量一般较小,所以repartition的时间短,不像spark_batch一次处理大量数据一旦repartition则会特别久,所以最终还是要根据具体情况测试来决定。

5 总结

将Receiver模式改成Direct模式,实现了资源优化,提升了程序的稳定性,缺点是需要自己管理offset,操作相对复杂。未来,个推将不断探索和优化Spark Streaming技术,发挥其强大的数据处理能力,为建设实时数仓提供保障。

收起阅读 »

android 插件开发,插件类 打开 activity ,然后接收 activity 返回值 onActivityResult

activity Android 5 SDK

也是郁闷了好些天才搞定

场景:如题,activity (页面)操作后返回数据到,插件类里面。
不说了,直接上代码:
插件类:

package com.example.chajian;  

import android.content.Intent;  
import android.util.Log;  

import org.json.JSONArray;  

import io.dcloud.common.DHInterface.IApp;  
import io.dcloud.common.DHInterface.ISysEventListener;  
import io.dcloud.common.DHInterface.IWebview;  
import io.dcloud.common.DHInterface.StandardFeature;  
import io.dcloud.common.util.JSUtil;  

public class Ceshi extends StandardFeature {  

    public void Test(IWebview pWebview, JSONArray array) {  

        final IWebview iWebview = pWebview;  
        final String CallBackID = array.optString(0);  
        final JSONArray newArray = new JSONArray();  

        newArray.put(array.optString(1));  
        newArray.put(array.optString(2));  
        newArray.put(array.optString(3));  
        newArray.put(array.optString(4));  

         //添加监听  
        final IApp _app = pWebview.obtainFrameView().obtainApp();  
        _app.registerSysEventListener(new ISysEventListener() {  
            @Override  
            public boolean onExecute(SysEventType pEventType, Object pArgs) {  

                Object[] _args = (Object[]) pArgs;  
                int requestCode = (Integer) _args[0];  
                int resultCode = (Integer) _args[1];  

                Intent data = (Intent) _args[2];  

                //[Ljava.lang.Object;@e89a7c8  
                Log.d("aaa", "onExecute: _args " + _args);  
                //1  
                Log.d("aaa", "onExecute: requestCode " + requestCode);  
                //-1  
                Log.d("aaa", "onExecute: resultCode " + resultCode);  
                //Intent { (has extras) }  
                Log.d("aaa", "onExecute: data " + data);  

                if (pEventType == SysEventType.onActivityResult) {  
                    //用完给取消注册监听  
                    _app.unregisterSysEventListener(this, SysEventType.onActivityResult);  

                    //判断请求码  
                    if (requestCode == 1) {  
                        //获取返回值  
                        String returnData = data.getStringExtra("return_data");  
                        //执行 js 回调  
                        JSUtil.execCallback(iWebview, CallBackID, returnData, JSUtil.OK, false);  
                        Log.d("aaa", "返回的数据为:" + returnData);  
                    }  
                }  
                return false;  
            }  
        }, SysEventType.onActivityResult);  

        //打开 activity  
        Intent openItent = new Intent(pWebview.getActivity(), Open.class);  
//        Intent openItent = new Intent(pWebview.getContext(),Open.class);  
        pWebview.getActivity().startActivityForResult(openItent, 1);  
    }  

}  

activity 类:

package com.example.chajian;  

import android.content.Intent;  
import android.os.Bundle;  
import android.support.v7.app.AppCompatActivity;  
import android.util.Log;  
import android.view.View;  
import android.widget.Button;  

public class Open extends AppCompatActivity {  

    @Override  
    protected void onCreate(Bundle savedInstanceState) {  
        super.onCreate(savedInstanceState);  
        setContentView(R.layout.open_activity);  

        Button open = findViewById(R.id.open);  

        open.setOnClickListener(new View.OnClickListener() {  
            @Override  
            public void onClick(View view) {  
                Log.d("aaa", "Open.onCreate : open");  

                Intent intent = new Intent();  
                intent.putExtra("return_data","我去它妹的!");  
                setResult(RESULT_OK,intent);  

                finish();  
            }  
        });  
    }  

}  

代码并不复杂,一行一行的看,很好理解,还有注释。
逻辑是,插件类设置事件监听-打开 activity-activity 设置返回值后 finish() (关闭)-插件类监听到返回值处理执行 js 回调。

继续阅读 »

也是郁闷了好些天才搞定

场景:如题,activity (页面)操作后返回数据到,插件类里面。
不说了,直接上代码:
插件类:

package com.example.chajian;  

import android.content.Intent;  
import android.util.Log;  

import org.json.JSONArray;  

import io.dcloud.common.DHInterface.IApp;  
import io.dcloud.common.DHInterface.ISysEventListener;  
import io.dcloud.common.DHInterface.IWebview;  
import io.dcloud.common.DHInterface.StandardFeature;  
import io.dcloud.common.util.JSUtil;  

public class Ceshi extends StandardFeature {  

    public void Test(IWebview pWebview, JSONArray array) {  

        final IWebview iWebview = pWebview;  
        final String CallBackID = array.optString(0);  
        final JSONArray newArray = new JSONArray();  

        newArray.put(array.optString(1));  
        newArray.put(array.optString(2));  
        newArray.put(array.optString(3));  
        newArray.put(array.optString(4));  

         //添加监听  
        final IApp _app = pWebview.obtainFrameView().obtainApp();  
        _app.registerSysEventListener(new ISysEventListener() {  
            @Override  
            public boolean onExecute(SysEventType pEventType, Object pArgs) {  

                Object[] _args = (Object[]) pArgs;  
                int requestCode = (Integer) _args[0];  
                int resultCode = (Integer) _args[1];  

                Intent data = (Intent) _args[2];  

                //[Ljava.lang.Object;@e89a7c8  
                Log.d("aaa", "onExecute: _args " + _args);  
                //1  
                Log.d("aaa", "onExecute: requestCode " + requestCode);  
                //-1  
                Log.d("aaa", "onExecute: resultCode " + resultCode);  
                //Intent { (has extras) }  
                Log.d("aaa", "onExecute: data " + data);  

                if (pEventType == SysEventType.onActivityResult) {  
                    //用完给取消注册监听  
                    _app.unregisterSysEventListener(this, SysEventType.onActivityResult);  

                    //判断请求码  
                    if (requestCode == 1) {  
                        //获取返回值  
                        String returnData = data.getStringExtra("return_data");  
                        //执行 js 回调  
                        JSUtil.execCallback(iWebview, CallBackID, returnData, JSUtil.OK, false);  
                        Log.d("aaa", "返回的数据为:" + returnData);  
                    }  
                }  
                return false;  
            }  
        }, SysEventType.onActivityResult);  

        //打开 activity  
        Intent openItent = new Intent(pWebview.getActivity(), Open.class);  
//        Intent openItent = new Intent(pWebview.getContext(),Open.class);  
        pWebview.getActivity().startActivityForResult(openItent, 1);  
    }  

}  

activity 类:

package com.example.chajian;  

import android.content.Intent;  
import android.os.Bundle;  
import android.support.v7.app.AppCompatActivity;  
import android.util.Log;  
import android.view.View;  
import android.widget.Button;  

public class Open extends AppCompatActivity {  

    @Override  
    protected void onCreate(Bundle savedInstanceState) {  
        super.onCreate(savedInstanceState);  
        setContentView(R.layout.open_activity);  

        Button open = findViewById(R.id.open);  

        open.setOnClickListener(new View.OnClickListener() {  
            @Override  
            public void onClick(View view) {  
                Log.d("aaa", "Open.onCreate : open");  

                Intent intent = new Intent();  
                intent.putExtra("return_data","我去它妹的!");  
                setResult(RESULT_OK,intent);  

                finish();  
            }  
        });  
    }  

}  

代码并不复杂,一行一行的看,很好理解,还有注释。
逻辑是,插件类设置事件监听-打开 activity-activity 设置返回值后 finish() (关闭)-插件类监听到返回值处理执行 js 回调。

收起阅读 »

【NEXT学院】uni-app商业级应用实战 一次搞定小程序/安卓/iOS/H5 2人拼团 ¥459.00

学习

【NEXT学院】uni-app商业级应用实战 一次搞定小程序/安卓/iOS/H5

2人拼团 ¥459.00

https://ke.qq.com/course/379043?group_id=403035#term_id=100451746

【NEXT学院】uni-app商业级应用实战 一次搞定小程序/安卓/iOS/H5

2人拼团 ¥459.00

https://ke.qq.com/course/379043?group_id=403035#term_id=100451746