通过AsyncTcpConnection循环访问https接口的json数据问题

use \GatewayWorker\Lib\Gateway;
use \Workerman\Connection\AsyncTcpConnection;
use \Workerman\Events\EventInterface;
use \Workerman\Worker;
use \Workerman\Autoloader;
use \Workerman\WebServer;
use \GatewayWorker\BusinessWorker;
use \Workerman\Lib\Timer;
use Lib\JsonFormat;
use Lib\JsonToSql;

include_once __DIR__.'/../'.'Include/function.php';
include_once __DIR__.'/../'.'Include/config.db.php';
require_once __DIR__.'/../../'.'vendor/workerman/mysql-master/src/Connection.php';

// bussinessWorker 进程
$worker = new BusinessWorker();
// worker名称
$worker->name = 'M-BtcTrade-COM';
// bussinessWorker进程数量
$worker->count = 1;
// 服务注册地址
$worker->registerAddress = '127.0.0.1:1238';

// 心跳间隔25秒
define('HEARTBEAT_TIME', 25);

$worker->onWorkerStart = function()
{

    // 设置访问对方主机的本地ip及端口以及ssl证书
    $context_option = array(
            // ssl选项，参考http://php.net/manual/zh/context.ssl.php
            'ssl' => array(
                    // 本地证书路径。 必须是 PEM 格式，并且包含本地的证书及私钥。
                    //'local_cert'        => '/your/path/to/pemfile',
                    // local_cert 文件的密码。
                    //'passphrase'        => 'your_pem_passphrase',
                    // 是否允许自签名证书。
                    'allow_self_signed' => true,
                    // 是否需要验证 SSL 证书。
                    'verify_peer'       => false,
                    'verify_peer_name'  => false
            )
    );
    // 每1秒执行一次
    $time_interval = 1;
    $connect_time = time();
    // 给connection对象临时添加一个timer_id属性保存定时器id
    Timer::add($time_interval, function()
    {
        $coin_list = array("btc","eth","ltc","doge","ybc");
        //print_r($coin_list);
        foreach ($coin_list as $key=>$val){
            $con = new AsyncTcpConnection("ssl://api.btctrade.com:443", $context_option);
            $con->cointype = $val;
            $con->send("GET /api/trades?coin=".$val." HTTP/1.1\r\nHost: api.btctrade.com\r\n\r\n");
            $con->onMessage = function($con, $data)
            {
                $data_arr = explode("\r\n\r\n", $data, 2);
                $data_arr1 = explode("\r\n", $data_arr);
                $json_data = "btctrade_trade_".$con->cointype."cny";
                $json_data = json_decode($data_arr1,true);
                //print_r($json_data);
                //echo $con->uri."\n";
                //输出数据进行处理
                ..................................................
                ................................................
            };
            $con->connect();

            $con->onClose = function($con)
            {
                //echo "connection closed\n";
            };

            //print_r($val);
            // 设置以ssl加密方式访问，使之成为wss
        }
    });

};

// 如果不是在根目录启动，则运行runAll方法
if(!defined('GLOBAL_START')) {
    Worker::runAll();
}

以上代码是每秒钟循环中抓取一下5个URL返回的JSON数据

现在的问题是在$con->onMessage还未完全处理完数据之前，$con 对象就被下次循环重建了，从而造成数据丢失，请问如何修改代码才能避免这个问题的出现呢？