资讯专栏INFORMATION COLUMN

微信开源mars源码分析2—上层samples分析(续)

MyFaith / 2326人阅读

摘要:本来是想直接深入到的核心层去看的,但是发现其实上面的部分还有好些没有分析到,因此回来继续分析。另外一个,是专用于统计的,我们暂时不去关注。具体的内容我会在后面的核心层分析的时候指出。准备下一篇进行的核心层分析吧。

本来是想直接深入到mars的核心层去看的,但是发现其实上面的samples部分还有好些没有分析到,因此回来继续分析。
ConversationActivity这个类中实际上还做了很多的工作,在onCreate中:

final MainService mainService = new MainService();
    MarsServiceProxy.setOnPushMessageListener(BaseConstants.CGIHISTORY_CMDID, mainService);
    MarsServiceProxy.setOnPushMessageListener(BaseConstants.CONNSTATUS_CMDID, mainService);
    MarsServiceProxy.setOnPushMessageListener(BaseConstants.FLOW_CMDID, mainService);
    MarsServiceProxy.setOnPushMessageListener(BaseConstants.PUSHMSG_CMDID, mainService);
    MarsServiceProxy.setOnPushMessageListener(BaseConstants.SDTRESULT_CMDID, mainService);

这里出现了一个MainService,我们来看看:
/mars-master/samples/android/marsSampleChat/app/src/main/java/com/tencent/mars/sample/core/MainService.java

public class MainService implements PushMessageHandler {
        
            public static String TAG = "Mars.Sample.MainService";
        
            private Thread recvThread;
        
            private LinkedBlockingQueue pushMessages = new LinkedBlockingQueue<>();
        
            private BusinessHandler[] handlers = new BusinessHandler[]{
                    new MessageHandler(),
                    new StatisticHandler()
            };
        
            public MainService() {
                this.start();
            }
        
            public void start() {
                if (recvThread == null) {
                    recvThread = new Thread(pushReceiver, "PUSH-RECEIVER");
        
                    recvThread.start();
                }
            }
        
            private final Runnable pushReceiver = new Runnable() {
                @Override
                public void run() {
                    while (true) {
                        try {
                            PushMessage pushMessage = pushMessages.take();
                            if (pushMessage != null) {
                                for (BusinessHandler handler : handlers) {
                                    if (handler.handleRecvMessage(pushMessage)) {
                                        break;
                                    }
                                }
                            }
        
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                            try {
                                Thread.sleep(500);
                            } catch (InterruptedException e1) {
                                //
                            }
                        }
                    }
                }
            };
        
            @Override
            public void process(PushMessage message) {
                pushMessages.offer(message);
            }
        }
1.启动了一个接受者线程pushReceiver;
2.pushReceiver从LinkedBlockingQueue的pushMessages中不断获取message,然后通知到handlers的每个成员中,handlers是这样定义的:
private BusinessHandler[] handlers = new BusinessHandler[]{
                new MessageHandler(),
                new StatisticHandler()
        };
继续往下看MessageHandler:
/mars-master/samples/android/marsSampleChat/app/src/main/java/com/tencent/mars/sample/core/MessageHandler.java
public class MessageHandler extends BusinessHandler{
    
        public static String TAG = MessageHandler.class.getSimpleName();
    
        @Override
        public boolean handleRecvMessage(PushMessage pushMessage) {
    
            switch (pushMessage.cmdId) {
                case Constants.PUSHCMD:
                {
                    try {
                        Messagepush.MessagePush message = Messagepush.MessagePush.parseFrom(pushMessage.buffer);
                        Intent intent = new Intent();
                        intent.setAction(Constants.PUSHACTION);
                        intent.putExtra("msgfrom", message.from);
                        intent.putExtra("msgcontent", message.content);
                        intent.putExtra("msgtopic", message.topic);
                        SampleApplicaton.getContext().sendBroadcast(intent);
                    } catch (InvalidProtocolBufferNanoException e) {
                        Log.e(TAG, "%s", e.toString());
                    }
                }
                    return true;
                default:
                    break;
            }
    
            return false;
        }
    }

如果是Constants.PUSHCMD类型的message,那么就发送一个广播。这个广播实际上会由Mars核心部分接收到,但是传递的intent参数其实没有意义,核心部分只是根据每次的pushcmd进行网络状态的检查而已,这些后话我们在分析核心的时候再说。
另外一个StatisticHandler,是专用于统计的,我们暂时不去关注。
让我们回到ConversationActivity,在MainService的new之后,会调用MarsServiceProxy.setOnPushMessageListener多次,设置监听,我们来看看MarsServiceProxy里面如何运转:

public static void setOnPushMessageListener(int cmdId, PushMessageHandler pushMessageHandler) {
    if (pushMessageHandler == null) {
        inst.pushMessageHandlerHashMap.remove(cmdId);
    } else {
        inst.pushMessageHandlerHashMap.put(cmdId, pushMessageHandler);
    }
}

添加了监听到一个支持高并发的hashmap中:

private ConcurrentHashMap pushMessageHandlerHashMap = new ConcurrentHashMap<>();
private MarsPushMessageFilter filter = new MarsPushMessageFilter.Stub() {

    @Override
    public boolean onRecv(int cmdId, byte[] buffer) throws RemoteException {
        PushMessageHandler handler = pushMessageHandlerHashMap.get(cmdId);
        if (handler != null) {
            Log.i(TAG, "processing push message, cmdid = %d", cmdId);
            PushMessage message = new PushMessage(cmdId, buffer);
            handler.process(message);
            return true;

        } else {
            Log.i(TAG, "no push message listener set for cmdid = %d, just ignored", cmdId);
        }

        return false;
    }
};

可以看到,这里创建了一个filter,这个filter在接收到一个cmd后,根据id查找到PushMessageHandler,然后调用hander.process,传递参数PushMessage,这个PushMessage实际上又是由cmdid和buffer组成的。这里的cmdid理解为一个指令的类型id即可。再到MainService中的process,其实就是加入PushMessage的一个队列中等待处理。
回来看filter在何时调用的吧。
首先在MarsServiceProxy的onServiceConnected:

@Override
public void onServiceConnected(ComponentName componentName, IBinder iBinder) {
    Log.d(TAG, "remote mars service connected");

    try {
        service = MarsService.Stub.asInterface(iBinder);
        service.registerPushMessageFilter(filter);
        service.setAccountInfo(accountInfo.uin, accountInfo.userName);

    } catch (Exception e) {
        service = null;
    }
}

看到了吧,将这个filter注册到了服务中。来吧,MarsServiceNative的registerPushMessageFilter:

@Override
public void registerPushMessageFilter(MarsPushMessageFilter filter) throws RemoteException {
    stub.registerPushMessageFilter(filter);
}

到了MarsServiceStub里:

@Override
public void registerPushMessageFilter(MarsPushMessageFilter filter) throws RemoteException {
    filters.remove(filter);
    filters.add(filter);
}

加入了filters的队列中,然后在onPush中有所调用:

@Override
public void onPush(int cmdid, byte[] data) {
    for (MarsPushMessageFilter filter : filters) {
        try {
            if (filter.onRecv(cmdid, data)) {
                break;
            }

        } catch (RemoteException e) {
            //
        }
    }
}

这里调用了每个filter的onRecv方法,这样就和上面的串起来了吧。那么何时调用的这个onPush呢,答案在核心mars部分的StnLogic里面,这里规定了一个ICallBack,里面有onPush,会在适当的时候调用。具体的内容我会在后面的mars核心层分析的时候指出。

总结一下:
MainService在一开始启动,并且启动专门的线程处理接收到的pushMessage,处理的过程就是调用之前已经准备好的一个handler的队列的每个项目的handleRecvMessage,其中的MessageHandler会发送广播通知mars有pushcmd来了。
另一方面,MainService提供process调用向message队列中加入新项目。这个加入的过程调用在MarsServiceProxy中进行,内部生成了一个filter过滤器,并将其注册到服务MarsServiceNative中,MarsServiceNative其实也是走的MarsServiceStub,在他里面维护的filter的队列,并且MarsServiceStub还是底层核心Mars的回调监听者,在onPush到来的时候,依次调用filter队列中的各个项目的onRecv方法来实现通知。在filter的onRecv方法中就实现了handler.process(message),通知到了handler。

现在对于这部分应该比较明晰了吧。准备下一篇进行Mars的核心层分析吧。

文章版权归作者所有,未经允许请勿转载,若此文章存在违规行为,您可以联系管理员删除。

转载请注明本文地址:https://www.ucloud.cn/yun/66977.html

相关文章

  • 微信开源mars源码分析1—上层samples分析

    摘要:微信已经开源了,但是市面上相关的文章较少,即使有也是多在于使用等这些,那么这次我希望能够从这个直接用于底层通讯的部分进行个分析。首先明确下,微信用了的开源协议库,来代替和。核心的部分我们先放下,下一篇再深入分析。 微信已经开源了mars,但是市面上相关的文章较少,即使有也是多在于使用xlog等这些,那么这次我希望能够从stn这个直接用于im底层通讯的部分进行个分析。为了能分析的全面些,...

    caiyongji 评论0 收藏0
  • 微信开源mars源码分析5—底层核心mars分析(2)

    摘要:执行并根据每个连接的状态决定后续处理,上篇已经讲过,不再累述。上面的三段处理完毕后,应该是数组中不再有连接才对,这里的保险处理是对数组再进行检查。至此跳出,算是整个连接过程完毕了。这里需要逐句分析,首先是。 最近回顾之前的文章,发现最后一篇有些着急了,很多地方没有叙述清楚。这里先做个衔接吧。我们还是以长连接为例,从longlink.cc看起。首先是那个线程函数__Run:/mars-m...

    asce1885 评论0 收藏0
  • 微信终端开源数据库 WCDB - Swift 版本

    摘要:作为微信的终端数据库,从开源至今,共迭代了个版本。微信也转向开发了吗相信这会是大家非常关心的问题。不仅微信,国内外大部分都还没有完全转向,但显然这是个趋势。另一方面,没有微信的上线机制的保护和庞大的用户量的验证,我们需要确保的稳定性。 WCDB 作为微信的终端数据库,从 2017.6 开源至今,共迭代了 5 个版本。我们一直关注开发者们的需求,并不断优化性能,新增如全文搜索等常用的功能...

    CloudDeveloper 评论0 收藏0

发表评论

0条评论

MyFaith

|高级讲师

TA的文章

阅读更多
最新活动
阅读需要支付1元查看
<