diff --git a/app/Amqp/Consumer/ChatMessageConsumer.php b/app/Amqp/Consumer/ChatMessageConsumer.php
index b4d7d92..3b3c714 100644
--- a/app/Amqp/Consumer/ChatMessageConsumer.php
+++ b/app/Amqp/Consumer/ChatMessageConsumer.php
@@ -4,21 +4,22 @@ declare(strict_types=1);
 
 namespace App\Amqp\Consumer;
 
+use App\Model\UsersFriend;
+use Hyperf\Amqp\Annotation\Consumer;
+use Hyperf\Amqp\Result;
+use Hyperf\Amqp\Message\ConsumerMessage;
+use Hyperf\Redis\Redis;
+use Hyperf\Amqp\Message\Type;
+use Hyperf\Amqp\Builder\QueueBuilder;
+use PhpAmqpLib\Message\AMQPMessage;
+use App\Model\User;
 use App\Helper\PushMessageHelper;
 use App\Model\Chat\ChatRecord;
 use App\Model\Chat\ChatRecordsCode;
 use App\Model\Chat\ChatRecordsFile;
-use App\Model\User;
 use App\Model\Chat\ChatRecordsInvite;
 use App\Service\SocketFDService;
 use App\Service\SocketRoomService;
-use Hyperf\Amqp\Result;
-use Hyperf\Amqp\Annotation\Consumer;
-use Hyperf\Amqp\Message\ConsumerMessage;
-use Hyperf\Redis\Redis;
-use PhpAmqpLib\Message\AMQPMessage;
-use Hyperf\Amqp\Message\Type;
-use Hyperf\Amqp\Builder\QueueBuilder;
 
 /**
  * @Consumer(name="聊天消息消费者",enable=true)
@@ -56,9 +57,17 @@ class ChatMessageConsumer extends ConsumerMessage
      */
     private $socketRoomService;
 
+    const EVENTS = [
+        'event_talk' => 'onConsumeTalk',
+        'event_keyboard' => 'onConsumeKeyboard',
+        'event_online_status' => 'onConsumeOnlineStatus',
+        'event_revoke_talk' => 'onConsumeRevokeTalk',
+    ];
+
     /**
      * ChatMessageConsumer constructor.
      * @param SocketFDService $socketFDService
+     * @param SocketRoomService $socketRoomService
      */
     public function __construct(SocketFDService $socketFDService, SocketRoomService $socketRoomService)
     {
@@ -87,6 +96,22 @@ class ChatMessageConsumer extends ConsumerMessage
      * @return string
      */
     public function consumeMessage($data, AMQPMessage $message): string
+    {
+        if (isset($data['event'])) {
+            return $this->{self::EVENTS[$data['event']]}($data, $message);
+        }
+
+        return Result::ACK;
+    }
+
+    /**
+     * 对话聊天消息
+     *
+     * @param array $data 队列消息
+     * @param AMQPMessage $message
+     * @return string
+     */
+    public function onConsumeTalk(array $data, AMQPMessage $message)
     {
         $redis = container()->get(Redis::class);
 
@@ -96,13 +121,13 @@ class ChatMessageConsumer extends ConsumerMessage
             return Result::ACK;
         }
 
-        $source = $data['source'];
+        $source = $data['data']['source'];
 
-        $fids = $this->socketFDService->findUserFds($data['sender']);
+        $fids = $this->socketFDService->findUserFds($data['data']['sender']);
         if ($source == 1) {// 私聊
-            $fids = array_merge($fids, $this->socketFDService->findUserFds($data['receive']));
+            $fids = array_merge($fids, $this->socketFDService->findUserFds($data['data']['receive']));
         } else if ($source == 2) {//群聊
-            $userIds = $this->socketRoomService->getRoomMembers(strval($data['receive']));
+            $userIds = $this->socketRoomService->getRoomMembers(strval($data['data']['receive']));
             foreach ($userIds as $uid) {
                 $fids = array_merge($fids, $this->socketFDService->findUserFds(intval($uid)));
             }
@@ -118,7 +143,7 @@ class ChatMessageConsumer extends ConsumerMessage
          * @var ChatRecord
          */
         $result = ChatRecord::leftJoin('users', 'users.id', '=', 'chat_records.user_id')
-            ->where('chat_records.id', $data['record_id'])
+            ->where('chat_records.id', $data['data']['record_id'])
             ->first([
                 'chat_records.id',
                 'chat_records.source',
@@ -133,6 +158,10 @@ class ChatMessageConsumer extends ConsumerMessage
                 'users.avatar as avatar',
             ]);
 
+        if (!$result) {
+            return Result::ACK;
+        }
+
         $file = [];
         $code_block = [];
         $forward = [];
@@ -171,9 +200,9 @@ class ChatMessageConsumer extends ConsumerMessage
         }
 
         $msg = [
-            'send_user' => $data['sender'],
-            'receive_user' => $data['receive'],
-            'source_type' => $data['source'],
+            'send_user' => $data['data']['sender'],
+            'receive_user' => $data['data']['receive'],
+            'source_type' => $data['data']['source'],
             'data' => PushMessageHelper::formatTalkMsg([
                 'id' => $result->id,
                 'msg_type' => $result->msg_type,
@@ -195,11 +224,99 @@ class ChatMessageConsumer extends ConsumerMessage
         foreach ($fids as $fd) {
             $fd = intval($fd);
             if ($server->exist($fd)) {
-                $server->push($fd, json_encode(['chat_message', $msg]));
+                $server->push($fd, json_encode(['event_talk', $msg]));
+            }
+        }
+
+        return Result::ACK;
+    }
+
+    /**
+     * 键盘输入事件消息
+     *
+     * @param array $data 队列消息
+     * @param AMQPMessage $message
+     * @return string
+     */
+    public function onConsumeKeyboard(array $data, AMQPMessage $message)
+    {
+        $fds = $this->socketFDService->findUserFds($data['data']['receive_user']);
+        $server = server();
+        foreach ($fds as $fd) {
+            $fd = intval($fd);
+            if ($server->exist($fd)) {
+                $server->push($fd, json_encode(['event_keyboard', $data['data']]));
+            }
+        }
+
+        return Result::ACK;
+    }
+
+    /**
+     * 用户上线或下线消息
+     *
+     * @param array $data 队列消息
+     * @param AMQPMessage $message
+     * @return string
+     */
+    public function onConsumeOnlineStatus(array $data, AMQPMessage $message)
+    {
+        $user_id = $data['data']['user_id'];
+        $friends = UsersFriend::getFriendIds(intval($user_id));
+
+        $fds = [];
+        foreach ($friends as $friend_id) {
+            $fds = array_merge($fds, $this->socketFDService->findUserFds(intval($friend_id)));
+        }
+
+        $fds = array_unique($fds);
+        $server = server();
+        foreach ($fds as $fd) {
+            $fd = intval($fd);
+            if ($server->exist($fd)) {
+                $server->push($fd, json_encode(['event_online_status', $data['data']]));
+            }
+        }
+
+        return Result::ACK;
+    }
+
+    /**
+     * 撤销聊天消息
+     *
+     * @param array $data 队列消息
+     * @param AMQPMessage $message
+     * @return string
+     */
+    public function onConsumeRevokeTalk(array $data, AMQPMessage $message)
+    {
+        $record = ChatRecord::where('id', $data['data']['record_id'])->first(['id', 'source', 'user_id', 'receive_id']);
+
+        $fds = [];
+        if ($record->source == 1) {
+            $fds = array_merge($fds, $this->socketFDService->findUserFds($record->user_id));
+            $fds = array_merge($fds, $this->socketFDService->findUserFds($record->receive_id));
+        } else if ($record->source == 2) {
+            $userIds = $this->socketRoomService->getRoomMembers(strval($record->receive_id));
+            foreach ($userIds as $uid) {
+                $fds = array_merge($fds, $this->socketFDService->findUserFds(intval($uid)));
+            }
+        }
+
+        $fds = array_unique($fds);
+        $server = server();
+        foreach ($fds as $fd) {
+            $fd = intval($fd);
+            if ($server->exist($fd)) {
+                $server->push($fd, json_encode(['event_revoke_talk', [
+                    'record_id' => $record->id,
+                    'source' => $record->source,
+                    'user_id' => $record->user_id,
+                    'receive_id' => $record->receive_id,
+                ]]));
             }
         }
 
-        unset($fids, $result, $msg);
         return Result::ACK;
     }
 }
diff --git a/app/Amqp/Producer/ChatMessageProducer.php b/app/Amqp/Producer/ChatMessageProducer.php
index 1a691bd..05091a3 100644
--- a/app/Amqp/Producer/ChatMessageProducer.php
+++ b/app/Amqp/Producer/ChatMessageProducer.php
@@ -14,14 +14,31 @@ class ChatMessageProducer extends ProducerMessage
 
     public $type = Type::FANOUT;
 
-    public function __construct($sender, $receive, $source, $record_id)
+    const EVENTS = [
+        'event_talk',
+        'event_keyboard',
+        'event_online_status',
+        'event_revoke_talk'
+    ];
+
+    /**
+     * 初始化处理...
+     *
+     * @param string $event 事件名
+     * @param array $data 数据
+     * @param array $options 其它参数
+     */
+    public function __construct(string $event, array $data, array $options = [])
     {
+        if (!in_array($event, self::EVENTS)) {
+            new \Exception('事件名未注册...');
+        }
+
         $message = [
-            'uuid' => $this->uuid(),
-            'sender' => intval($sender),  //发送者ID
-            'receive' => intval($receive),  //接收者ID
-            'source' => intval($source), //接收者类型 1:好友;2:群组
-            'record_id' => intval($record_id)
+            'uuid' => $this->uuid(),// 自定义消息ID
+            'event' => $event,
+            'data' => $data,
+            'options' => $options
         ];
 
         $this->payload = $message;
diff --git a/app/Controller/Api/V1/GroupController.php b/app/Controller/Api/V1/GroupController.php
index 64674e8..e960921 100644
--- a/app/Controller/Api/V1/GroupController.php
+++ b/app/Controller/Api/V1/GroupController.php
@@ -75,7 +75,12 @@ class GroupController extends CController
 
         // ...消息推送队列
         $this->producer->produce(
-            new ChatMessageProducer($user_id, $data['group_id'], 2, $data['record_id'])
+            new ChatMessageProducer('event_talk', [
+                'sender' => $user_id,  //发送者ID
+                'receive' => intval($data['group_id']),  //接收者ID
+                'source' => 2, //接收者类型 1:好友;2:群组
+                'record_id' => intval($data['record_id'])
+            ])
         );
 
         return $this->response->success([
@@ -134,7 +139,12 @@ class GroupController extends CController
 
         // ...消息推送队列
         $this->producer->produce(
-            new ChatMessageProducer($user_id, $params['group_id'], 2, $record_id)
+            new ChatMessageProducer('event_talk', [
+                'sender' => $user_id,  //发送者ID
+                'receive' => intval($params['group_id']),  //接收者ID
+                'source' => 2, //接收者类型 1:好友;2:群组
+                'record_id' => $record_id
+            ])
         );
 
         return $this->response->success([], '好友已成功加入群聊...');
@@ -163,7 +173,12 @@ class GroupController extends CController
 
         // ...消息推送队列
         $this->producer->produce(
-            new ChatMessageProducer($user_id, $params['group_id'], 2, $record_id)
+            new ChatMessageProducer('event_talk', [
+                'sender' => $user_id,  //发送者ID
+                'receive' => intval($params['group_id']),  //接收者ID
+                'source' => 2, //接收者类型 1:好友;2:群组
+                'record_id' => $record_id
+            ])
         );
 
         return $this->response->success([], '已成功退出群组...');
@@ -224,7 +239,12 @@ class GroupController extends CController
 
         // ...消息推送队列
         $this->producer->produce(
-            new ChatMessageProducer($user_id, $params['group_id'], 2, $record_id)
+            new ChatMessageProducer('event_talk', [
+                'sender' => $user_id,  //发送者ID
+                'receive' => intval($params['group_id']),  //接收者ID
+                'source' => 2, //接收者类型 1:好友;2:群组
+                'record_id' => $record_id
+            ])
         );
 
         return $this->response->success([], '已成功退出群组...');
diff --git a/app/Controller/Api/V1/TalkController.php b/app/Controller/Api/V1/TalkController.php
index 150e836..eac6dd2 100644
--- a/app/Controller/Api/V1/TalkController.php
+++ b/app/Controller/Api/V1/TalkController.php
@@ -227,6 +227,14 @@ class TalkController extends CController
         ]);
 
         [$isTrue, $message,] = $this->talkService->revokeRecord($this->uid(), $params['record_id']);
+        if($isTrue){
+            $this->producer->produce(
+                new ChatMessageProducer('event_revoke_talk', [
+                    'record_id' => $params['record_id']
+                ])
+            );
+        }
+
 
         return $isTrue
             ? $this->response->success([], $message)
@@ -323,7 +331,12 @@ class TalkController extends CController
         // ...消息推送队列
         foreach ($ids as $value) {
             $this->producer->produce(
-                new ChatMessageProducer($user_id, $value['receive_id'], $value['source'], $value['record_id'])
+                new ChatMessageProducer('event_talk', [
+                    'sender' => $user_id,  //发送者ID
+                    'receive' => intval($value['receive_id']),  //接收者ID
+                    'source' => intval($value['source']), //接收者类型 1:好友;2:群组
+                    'record_id' => $value['record_id']
+                ])
             );
         }
 
@@ -518,7 +531,12 @@ class TalkController extends CController
 
         // ...消息推送队列
         $this->producer->produce(
-            new ChatMessageProducer($user_id, $params['receive_id'], $params['source'], $record_id)
+            new ChatMessageProducer('event_talk', [
+                'sender' => $user_id,  //发送者ID
+                'receive' => intval($params['receive_id']),  //接收者ID
+                'source' => intval($params['source']), //接收者类型 1:好友;2:群组
+                'record_id' => $record_id
+            ])
         );
 
         return $this->response->success();
@@ -558,7 +576,12 @@ class TalkController extends CController
 
         // ...消息推送队列
         $this->producer->produce(
-            new ChatMessageProducer($user_id, $params['receive_id'], $params['source'], $record_id)
+            new ChatMessageProducer('event_talk', [
+                'sender' => $user_id,  //发送者ID
+                'receive' => intval($params['receive_id']),  //接收者ID
+                'source' => intval($params['source']), //接收者类型 1:好友;2:群组
+                'record_id' => $record_id
+            ])
         );
 
         return $this->response->success();
@@ -615,7 +638,12 @@ class TalkController extends CController
 
         // ...消息推送队列
         $this->producer->produce(
-            new ChatMessageProducer($user_id, $params['receive_id'], $params['source'], $record_id)
+            new ChatMessageProducer('event_talk', [
+                'sender' => $user_id,  //发送者ID
+                'receive' => intval($params['receive_id']),  //接收者ID
+                'source' => intval($params['source']), //接收者类型 1:好友;2:群组
+                'record_id' => $record_id
+            ])
         );
 
 
@@ -668,7 +696,12 @@ class TalkController extends CController
 
         // ...消息推送队列
         $this->producer->produce(
-            new ChatMessageProducer($user_id, $params['receive_id'], $params['source'], $record_id)
+            new ChatMessageProducer('event_talk', [
+                'sender' => $user_id,  //发送者ID
+                'receive' => intval($params['receive_id']),  //接收者ID
+                'source' => intval($params['source']), //接收者类型 1:好友;2:群组
+                'record_id' => $record_id
+            ])
         );
 
         return $this->response->success();
diff --git a/app/Controller/WebSocketController.php b/app/Controller/WebSocketController.php
index 7ab11b8..4f17feb 100644
--- a/app/Controller/WebSocketController.php
+++ b/app/Controller/WebSocketController.php
@@ -3,11 +3,6 @@ declare(strict_types=1);
 
 namespace App\Controller;
 
-use App\Cache\LastMsgCache;
-use App\Cache\UnreadTalkCache;
-use App\Model\Chat\ChatRecord;
-use App\Model\Group\UsersGroupMember;
-use App\Service\SocketRoomService;
 use Hyperf\Di\Annotation\Inject;
 use Hyperf\Contract\OnCloseInterface;
 use Hyperf\Contract\OnMessageInterface;
@@ -16,11 +11,14 @@ use Phper666\JWTAuth\JWT;
 use Swoole\Http\Request;
 use Swoole\Websocket\Frame;
 use Hyperf\Amqp\Producer;
-use App\Amqp\Producer\ChatMessageProducer;
 use Swoole\Http\Response;
 use Swoole\WebSocket\Server;
 use App\Traits\WebSocketTrait;
 use App\Service\SocketFDService;
+use App\Service\MessageHandleService;
+use App\Service\SocketRoomService;
+use App\Model\Group\UsersGroupMember;
+use App\Amqp\Producer\ChatMessageProducer;
 
 /**
  * Class WebSocketController
@@ -54,6 +52,17 @@ class WebSocketController implements OnMessageInterface, OnOpenInterface, OnClos
      */
     private $socketRoomService;
 
+    /**
+     * @inject
+     * @var MessageHandleService
+     */
+    private $messageHandleService;
+
+    const EVENTS = [
+        'event_talk' => 'onTalk',
+        'event_keyboard' => 'onKeyboard',
+    ];
+
     /**
      * 连接创建成功回调事件
      *
@@ -66,6 +75,9 @@ class WebSocketController implements OnMessageInterface, OnOpenInterface, OnClos
         $userInfo = $this->jwt->getParserData($token);
         stdout_log()->notice("用户连接信息 : user_id:{$userInfo['user_id']} | fd:{$request->fd} 时间：" . date('Y-m-d H:i:s'));
 
+        // 判断是否存在异地登录
+        $isOnline = $this->socketFDService->isOnlineAll(intval($userInfo['user_id']));
+
         // 绑定fd与用户关系
         $this->socketFDService->bindRelation($request->fd, $userInfo['user_id']);
 
@@ -74,6 +86,17 @@ class WebSocketController implements OnMessageInterface, OnOpenInterface, OnClos
         foreach ($groupIds as $group_id) {
             $this->socketRoomService->addRoomMember($userInfo['user_id'], $group_id);
         }
+
+        if (!$isOnline) {
+            // 推送消息至队列
+            $this->producer->produce(
+                new ChatMessageProducer('event_online_status', [
+                    'user_id' => $userInfo['user_id'],
+                    'status' => 1,
+                    'notify' => '好友上线通知...'
+                ])
+            );
+        }
     }
 
     /**
@@ -87,59 +110,12 @@ class WebSocketController implements OnMessageInterface, OnOpenInterface, OnClos
         // 判断是否为心跳检测
         if ($frame->data == 'PING') return;
 
-        // 当前用户ID
-        $user_id = $this->socketFDService->findFdUserId($frame->fd);
-
         [$event, $data] = array_values(json_decode($frame->data, true));
-        if ($user_id != $data['send_user']) {
+
+        if (isset(self::EVENTS[$event])) {
+            $this->messageHandleService->{self::EVENTS[$event]}($server, $frame, $data);
             return;
         }
-
-        //验证消息类型 私聊|群聊
-        if (!in_array($data['source_type'], [1, 2])) {
-            return;
-        }
-
-        //验证发送消息用户与接受消息用户之间是否存在好友或群聊关系(后期走缓存)
-//        if ($data['source_type'] == 1) {//私信
-//            //判断发送者和接受者是否是好友关系
-//            if (!UsersFriend::isFriend(intval($data['send_user']), intval($data['receive_user']))) {
-//                return;
-//            }
-//        } else if ($data['source_type'] == 2) {//群聊
-//            //判断是否属于群成员
-//            if (!UsersGroup::isMember(intval($data['receive_user']), intval($data['send_user']))) {
-//                return;
-//            }
-//        }
-
-        $result = ChatRecord::create([
-            'source' => $data['source_type'],
-            'msg_type' => 1,
-            'user_id' => $data['send_user'],
-            'receive_id' => $data['receive_user'],
-            'content' => htmlspecialchars($data['text_message']),
-            'created_at' => date('Y-m-d H:i:s'),
-        ]);
-
-        if (!$result) return;
-
-        // 判断是否私聊
-        if ($data['source_type'] == 1) {
-            $msg_text = mb_substr($result->content, 0, 30);
-            // 缓存最后一条消息
-            LastMsgCache::set([
-                'text' => $msg_text,
-                'created_at' => $result->created_at
-            ], intval($data['receive_user']), intval($data['send_user']));
-
-            // 设置好友消息未读数
-            make(UnreadTalkCache::class)->setInc(intval($result->receive_id), strval($result->user_id));
-        }
-
-        $this->producer->produce(
-            new ChatMessageProducer($data['send_user'], $data['receive_user'], $data['source_type'], $result->id)
-        );
     }
 
     /**
@@ -153,18 +129,23 @@ class WebSocketController implements OnMessageInterface, OnOpenInterface, OnClos
     {
         $user_id = $this->socketFDService->findFdUserId($fd);
 
-        stdout_log()->notice("客户端FD:{$fd} 已关闭连接 ，用户ID为【{$user_id}】，关闭时间：" . date('Y-m-d H:i:s'));
+        // stdout_log()->notice("客户端FD:{$fd} 已关闭连接 ，用户ID为【{$user_id}】，关闭时间：" . date('Y-m-d H:i:s'));
 
         // 解除fd关系
         $this->socketFDService->removeRelation($fd);
 
         // 判断是否存在异地登录
         $isOnline = $this->socketFDService->isOnlineAll(intval($user_id));
+        // ... 不存在异地登录，推送下线通知消息
         if (!$isOnline) {
-            // ... 不存在异地登录，推送下线通知消息
-            // ... 包装推送消息至队列
-        } else {
-            stdout_log()->notice("用户:{$user_id} 存在异地登录...");
+            // 推送消息至队列
+            $this->producer->produce(
+                new ChatMessageProducer('event_online_status', [
+                    'user_id' => $user_id,
+                    'status' => 0,
+                    'notify' => '好友离线通知通知...'
+                ])
+            );
         }
     }
 }
diff --git a/app/Helper/JsonHelper.php b/app/Helper/JsonHelper.php
new file mode 100644
index 0000000..eab9dc2
--- /dev/null
+++ b/app/Helper/JsonHelper.php
@@ -0,0 +1,10 @@
+<?php
+
+
+namespace App\Helper;
+
+
+class JsonHelper
+{
+
+}
diff --git a/app/Service/GroupService.php b/app/Service/GroupService.php
index 9cdbdbd..de15ce7 100644
--- a/app/Service/GroupService.php
+++ b/app/Service/GroupService.php
@@ -266,6 +266,8 @@ class GroupService extends BaseService
                     throw new Exception('添加群通知记录失败2  : quitGroupChat');
                 }
 
+                UsersChatList::where('uid', $user_id)->where('group_id', $group_id)->update(['status' => 0, 'updated_at' => date('Y-m-d H:i:s')]);
+
                 $record_id = $result->id;
             }
 
@@ -322,6 +324,10 @@ class GroupService extends BaseService
                 throw new Exception('添加群通知记录失败2');
             }
 
+            foreach ($member_ids as $member_id) {
+                UsersChatList::where('uid', $member_id)->where('group_id', $group_id)->update(['status' => 0, 'updated_at' => date('Y-m-d H:i:s')]);
+            }
+
             Db::commit();
         } catch (Exception $e) {
             Db::rollBack();
diff --git a/app/Service/MessageHandleService.php b/app/Service/MessageHandleService.php
new file mode 100644
index 0000000..e2d0e64
--- /dev/null
+++ b/app/Service/MessageHandleService.php
@@ -0,0 +1,122 @@
+<?php
+
+
+namespace App\Service;
+
+use Hyperf\Di\Annotation\Inject;
+use App\Amqp\Producer\ChatMessageProducer;
+use App\Cache\LastMsgCache;
+use App\Cache\UnreadTalkCache;
+use App\Model\Chat\ChatRecord;
+use App\Model\Group\UsersGroup;
+use App\Model\UsersFriend;
+use Hyperf\Amqp\Producer;
+use Swoole\Http\Response;
+use Swoole\WebSocket\Frame;
+use Swoole\WebSocket\Server;
+
+class MessageHandleService
+{
+    /**
+     * @Inject
+     * @var Producer
+     */
+    private $producer;
+
+    /**
+     * @inject
+     * @var SocketFDService
+     */
+    private $socketFDService;
+
+    /**
+     * 对话消息
+     *
+     * @param Response|Server $server
+     * @param Frame $frame
+     * @param array|string $data 解析后数据
+     * @return bool|void
+     */
+    public function onTalk($server, Frame $frame, $data)
+    {
+        // 当前用户ID
+        $user_id = $this->socketFDService->findFdUserId($frame->fd);
+
+        if ($user_id != $data['send_user']) {
+            return;
+        }
+
+        //验证消息类型 私聊|群聊
+        if (!in_array($data['source_type'], [1, 2])) {
+            return;
+        }
+
+        //验证发送消息用户与接受消息用户之间是否存在好友或群聊关系(后期走缓存)
+        if ($data['source_type'] == 1) {//私信
+            //判断发送者和接受者是否是好友关系
+            if (!UsersFriend::isFriend(intval($data['send_user']), intval($data['receive_user']))) {
+                return;
+            }
+        } else if ($data['source_type'] == 2) {//群聊
+            //判断是否属于群成员
+            if (!UsersGroup::isMember(intval($data['receive_user']), intval($data['send_user']))) {
+                return;
+            }
+        }
+
+        $result = ChatRecord::create([
+            'source' => $data['source_type'],
+            'msg_type' => 1,
+            'user_id' => $data['send_user'],
+            'receive_id' => $data['receive_user'],
+            'content' => htmlspecialchars($data['text_message']),
+            'created_at' => date('Y-m-d H:i:s'),
+        ]);
+
+        if (!$result) return;
+
+        // 判断是否私聊
+        if ($data['source_type'] == 1) {
+            $msg_text = mb_substr($result->content, 0, 30);
+            // 缓存最后一条消息
+            LastMsgCache::set([
+                'text' => $msg_text,
+                'created_at' => $result->created_at
+            ], intval($data['receive_user']), intval($data['send_user']));
+
+            // 设置好友消息未读数
+            make(UnreadTalkCache::class)->setInc(intval($result->receive_id), strval($result->user_id));
+        }
+
+        $this->producer->produce(
+            new ChatMessageProducer('event_talk', [
+                'sender' => intval($data['send_user']),  //发送者ID
+                'receive' => intval($data['receive_user']),  //接收者ID
+                'source' => intval($data['source_type']), //接收者类型 1:好友;2:群组
+                'record_id' => $result->id
+            ])
+        );
+
+        return true;
+    }
+
+    /**
+     * 键盘输入消息
+     *
+     * @param Response|Server $server
+     * @param Frame $frame
+     * @param array|string $data 解析后数据
+     * @return bool|void
+     */
+    public function onKeyboard($server, Frame $frame, $data)
+    {
+        $this->producer->produce(
+            new ChatMessageProducer('event_keyboard', [
+                'send_user' => intval($data['send_user']),  //发送者ID
+                'receive_user' => intval($data['receive_user']),  //接收者ID
+            ])
+        );
+
+        return true;
+    }
+}
diff --git a/app/Service/TalkService.php b/app/Service/TalkService.php
index 2570148..9354760 100644
--- a/app/Service/TalkService.php
+++ b/app/Service/TalkService.php
@@ -4,6 +4,7 @@ namespace App\Service;
 
 use App\Cache\FriendRemarkCache;
 use App\Cache\LastMsgCache;
+use App\Cache\UnreadTalkCache;
 use App\Model\Chat\ChatRecord;
 use App\Model\Chat\ChatRecordsCode;
 use App\Model\Chat\ChatRecordsFile;
@@ -48,7 +49,11 @@ class TalkService extends BaseService
             return [];
         }
 
-        $rows = array_map(function ($item) use ($user_id) {
+
+        $socketFDService = make(SocketFDService::class);
+        $runIdAll = $socketFDService->getServerRunIdAll();
+
+        $rows = array_map(function ($item) use ($user_id, $socketFDService, $runIdAll) {
             $data['id'] = $item['id'];
             $data['type'] = $item['type'];
             $data['friend_id'] = $item['friend_id'];
@@ -66,14 +71,13 @@ class TalkService extends BaseService
             if ($item['type'] == 1) {
                 $data['name'] = $item['nickname'];
                 $data['avatar'] = $item['user_avatar'];
-//                $data['unread_num'] = app('unread.talk')->get($user_id, $item['friend_id']);
-//                $data['online'] = app('client.manage')->isOnline($item['friend_id']);
+                $data['unread_num'] = make(UnreadTalkCache::class)->get($user_id, $item['friend_id']);
+                $data['online'] = $socketFDService->isOnlineAll($item['friend_id'], $runIdAll);
 
                 $remark = FriendRemarkCache::get($user_id, $item['friend_id']);
                 if ($remark) {
                     $data['remark_name'] = $remark;
                 } else {
-
                     $info = UsersFriend::select('user1', 'user2', 'user1_remark', 'user2_remark')
                         ->where('user1', ($user_id < $item['friend_id']) ? $user_id : $item['friend_id'])
                         ->where('user2', ($user_id < $item['friend_id']) ? $item['friend_id'] : $user_id)->first();
@@ -89,7 +93,6 @@ class TalkService extends BaseService
             }
 
             $records = LastMsgCache::get($item['type'] == 1 ? $item['friend_id'] : $item['group_id'], $item['type'] == 1 ? $user_id : 0);
-
             if ($records) {
                 $data['msg_text'] = $records['text'];
                 $data['updated_at'] = $records['created_at'];