提交推送服务

Pu Zhibing

1 天以前 915c265ef913601c2698faaa428f96f1e09a2369

 MessagePushTravel/src/main/java/com/sinata/push/util/echo/NettyServerController.java

@@ -10,16 +10,13 @@
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
import org.springframework.web.client.RestTemplate;

import javax.annotation.Resource;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.TimeUnit;
@@ -31,13 +28,16 @@
 * @createDate 2016年6月3日
 * @version 1.0
 */
@Component
public class NettyServerController {
   
   public static Hashtable<String, Hashtable<ChannelHandlerContext, String>> map = new Hashtable<String, Hashtable<ChannelHandlerContext,String>>();

   public static Hashtable<String,String> table;

   private RedisTemplate<String, String> redisTemplate = SpringUtil.getObject(StringRedisTemplate.class);
	
   @Resource
   private RedisTemplate<String, Object> redisTemplate;
	

   private SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");

@@ -99,7 +99,7 @@
            String device = jsonCon.getString("device");
            String version = jsonCon.getString("version");
                if(StringUtil.isNotEmpty(userId1)){
               String fluid_control = redisTemplate.opsForValue().get("fluid_control_" + userId1 + "_" + type);
               String fluid_control = (String)redisTemplate.opsForValue().get("fluid_control_" + userId1 + "_" + type);
               if(!StringUtils.hasLength(fluid_control)){
                  redisTemplate.opsForValue().set("fluid_control_" + userId1 + "_" + type, System.currentTimeMillis() + "");
               }else{
@@ -115,6 +115,13 @@

               //判断用户或者司机长连接
                    if(type==1){
                       //存储通讯通道
                       if(null != ctx && ctx.channel().isActive()){
                          System.err.println("开始存储用户通道：" + sdf.format(new Date()) + "----" + userId1);
                          NettyChannelMap.update("USER" + userId1, ctx);
                          String s = NettyMsg.setMsg(Method.ok, new HashMap<String, Object>());
                          ctx.writeAndFlush(Unpooled.copiedBuffer((s).getBytes()));
                       }
                        //确保账号在单个设备上登录
                        if(StringUtil.isNotEmpty(token)){
                            String token_ = (String)redisTemplate.opsForValue().get("USER_" + userId1);//获取缓存中最新的数据
@@ -140,57 +147,14 @@
                                redisTemplate.opsForValue().set("USER_" + userId1, token);
                            }
                        }
                  //存储通讯通道
                  if(null != ctx && ctx.channel().isActive()){
                     System.err.println("开始存储用户通道：" + sdf.format(new Date()) + "----" + userId1);
                     NettyChannelMap.update("USER" + userId1, ctx);
                     String s = NettyMsg.setMsg(Method.ok, new HashMap<String, Object>());
                     ctx.writeAndFlush(Unpooled.copiedBuffer((s).getBytes()));
                  }
                    }else{
                       Map<String, Object> params = new HashMap<>();
                  params.put("driverId", userId1);
                       HttpRequest post = HttpUtil.createPost(URLUtil.zuul + "/driver-server/base/driverOnline/addDriverOnline");
                       post.contentType(MediaType.APPLICATION_FORM_URLENCODED.getType());
                       post.form(params);
                       HttpResponse execute = post.execute();
                  if(200 != execute.getStatus()){
                     System.err.println("调用driver-server添加司机在线数据出错了");
                  }
                       JSONObject jsonObject = JSON.parseObject(execute.body(), JSONObject.class);
                  if(jsonObject.getIntValue("code") != 200){
                     System.err.println("调用driver-server添加司机在线数据出错了");
                  }

                  //TODO 存储最后一次上传的时间（用于保证车载端断电后1小时自动下班）
                  if(StringUtil.isNotEmpty(device) && device.equals("carDevice")){
                     redisTemplate.opsForValue().set("DEVICE_" + userId1, String.valueOf(System.currentTimeMillis()));

                     String token_ = (String)redisTemplate.opsForValue().get("DRIVER_" + userId1);//缓存中拿最新数据
                     if(StringUtil.isNotEmpty(token_) && !token_.equals(token)){
                        //如果是车载端登录，则将其它端都强迫下线
                        JSONObject msg_ = new JSONObject();
                        msg_.put("code", 200);
                        msg_.put("msg", "SUCCESS");
                        msg_.put("method", "OFFLINE");
                        msg_.put("data", new Object());
                        this.sendMsgToClient(ctx, msg_.toJSONString());//给当前通道发送消息
                        TimerTask timerTask = new TimerTask() {
                           @Override
                           public void run() {
                              NettyChannelMap.remove(ctx);
                           }
                        };
                        Timer timer = new Timer();
                        timer.schedule(timerTask, 3000);
                        timer.cancel();
                     }
                     if(StringUtil.isEmpty(token_)){//确保登录的时候存储token失败的情况
                        redisTemplate.opsForValue().set("DRIVER_" + userId1, token);
                     }
                  }


                       //存储通讯通道
                       if(null != ctx && ctx.channel().isActive()){
                          System.err.println("开始存储司机通道：" + sdf.format(new Date()) + "----" + userId1);
                          NettyChannelMap.update("DRIVER" + userId1, ctx);
                          String s = NettyMsg.setMsg(Method.ok, new HashMap<String, Object>());
                          ctx.writeAndFlush(Unpooled.copiedBuffer((s).getBytes()));
                       }
                        //确保账号在单个设备上登录
                  String value = (String)redisTemplate.opsForValue().get("DEVICE_" + userId1);
                  if(StringUtil.isNotEmpty(token) && StringUtil.isEmpty(value)){//APP端登录的操作
@@ -217,15 +181,6 @@
                                redisTemplate.opsForValue().set("DRIVER_" + userId1, token);
                            }
                        }


                        //存储通讯通道
                        if(null != ctx && ctx.channel().isActive()){
                            System.err.println("开始存储司机通道：" + sdf.format(new Date()) + "----" + userId1);
                            NettyChannelMap.update("DRIVER" + userId1, ctx);
                            String s = NettyMsg.setMsg(Method.ok, new HashMap<String, Object>());
                            ctx.writeAndFlush(Unpooled.copiedBuffer((s).getBytes()));
                        }
                    }
                }
         }
@@ -249,7 +204,7 @@
            Double lat = jsonCon.getDouble("lat");
            Double computeAzimuth = jsonCon.getDouble("computeAzimuth");
            Double altitude = jsonCon.getDouble("altitude");
            System.out.println("司机上传位置：" + sdf.format(new Date()) + "----" + driverId);
            System.out.println("司机上传位置：" + sdf.format(new Date()) + "----" + jsonCon.toJSONString());
            if(SinataUtil.isNotEmpty(driverId)){
               if(null !=  lon && 0 != lon && null !=  lat && 0 != lat){
                  if(null != orderId && 0 != driverId && null != orderType && 0 != orderType){//开始存入数据库
@@ -262,7 +217,7 @@
                     params.put("directionAngle", String.valueOf(computeAzimuth));
                     params.put("altitude", String.valueOf(altitude));
                     HttpRequest post = HttpUtil.createPost(URLUtil.zuul + "/driver-server/base/savePosition");
                     post.contentType(MediaType.APPLICATION_FORM_URLENCODED.getType());
                     post.contentType(MediaType.APPLICATION_FORM_URLENCODED_VALUE);
                     post.form(params);
                     HttpResponse execute = post.execute();
                     if(200 != execute.getStatus()){
@@ -273,19 +228,18 @@
                        System.err.println("调用driver-server存储位置数据出错了");
                     }
                  }
                  System.out.println("id:" + driverId + "---lon" + lon + "---lat" + lat);
                        redisTemplate.opsForValue().set("DRIVER" + driverId, lon + "," + lat, 300, TimeUnit.SECONDS);//实时位置存入redis中
               }else{
                  NettyServerController.sendMsgToClient(ctx, "__error__" + msg.toString());
                  this.sendMsgToClient(ctx, "__error__" + msg.toString());
               }
            }else{
               NettyServerController.sendMsgToClient(ctx, "__error__" + msg.toString());
               this.sendMsgToClient(ctx, "__error__" + msg.toString());
            }
         }
         
      } catch (Exception e) {
         if(isdebug) {
            NettyServerController.sendMsgToClient(ctx, "__error__" + msg.toString());
            this.sendMsgToClient(ctx, "__error__" + msg.toString());
         }
         e.printStackTrace();
      }
@@ -298,7 +252,7 @@
    * @param msg
    * @author TaoNingBo
    */
   public static void sendMsgToClient(ChannelHandlerContext ctx, String msg) {
   public void sendMsgToClient(ChannelHandlerContext ctx, String msg) {
      if (ctx != null && ctx.channel().isActive()) {
         ByteBuf buffer = Unpooled.copiedBuffer((msg).getBytes());
         ChannelFuture sync;