缓存方式幂等性工具、消息队列方式幂等性工具、检测客户端服务端是否在线-蒲公英云

1、缓存方式幂等性工具

import cn.hutool.core.util.IdUtil;
import com.alibaba.fastjson.JSON;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import java.util.concurrent.TimeUnit;
/** * 通过唯一ID判断数据是否存在 * 主要逻辑 存入数据不通过mysql主键进行做唯一标识， * 采用redis天然的幂等性，插入数据进行存储唯一标识，如果插入数据进行判断存在，不进行插入数据 */
public class IdentityHash { 
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    /** * 将对象存入缓存中 * @param obj 对象数据 * @param timeout 超时时间 */
    public void set(Object obj, long timeout) { 
        if (obj instanceof String) { 
            stringRedisTemplate.opsForValue().set(JSON.toJSONString(System.identityHashCode(obj)), IdUtil.randomUUID(), timeout, TimeUnit.MILLISECONDS);
            return;
        }
        stringRedisTemplate.opsForValue().set(JSON.toJSONString(System.identityHashCode(obj)), IdUtil.randomUUID(), timeout, TimeUnit.MILLISECONDS);
    }
    /** * 根据Key查询缓存中的数据 * @param key * @return */
    public String get(final String key) { 
        if (StringUtils.isEmpty(key)) { 
            return null;
        }
        return stringRedisTemplate.opsForValue().get(key);
    }
    /** * 根据key删除缓存数据 * @param key */
    public void delete(String key) { 
        stringRedisTemplate.delete(key);
    }
    /** * 查询缓存是否存在 * @param checkCase * @return */
    public boolean checkForm(String checkCase){ 
        String cacheValue = get(checkCase);
        /**如果查询缓存不为空，返回true*/
        if (StringUtils.isNotEmpty(cacheValue)){ 
            return true;
        }
        return false;
    }
}

2、消息队列方式幂等性工具

import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.io.IOException;
@Slf4j
@Service
public class AmqpHash { 
    @Autowired
    private AmqpTemplate amqpTemplate;
    /** * 发送消息 * * @param queueName 队列名称 */
    public void send(String queueName, Object obj) { 
        Messages messages = new Messages();
        messages.setObj(obj);
        messages.setMessageId(String.valueOf(System.identityHashCode(obj)));
        amqpTemplate.convertAndSend(queueName, JSON.toJSONString(messages));
    }
    @RabbitHandler
    @RabbitListener(queues = "queue", concurrency = "1")
    public void process(String msg, Message message, Channel channel) throws IOException { 
        log.debug("消费者接收到的消息 :{}, 来源：{} 时间：{}", msg, channel.getChannelNumber(), System.currentTimeMillis());
        try { 
            if(msg.contains("5")){ 
                throw new RuntimeException("抛出异常");
            }
            log.info("消息{}消费成功",msg);
            //消息Id
            message.getMessageProperties().getMessageId();
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
        } catch (Exception e) { 
            log.error("接收消息过程中出现异常，执行nack");
            //第三个参数为true表示异常消息重新返回队列，会导致一直在刷新消息，且返回的消息处于队列头部，影响后续消息的处理
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
            log.error("消息{}异常",message.getMessageProperties().getHeaders());
        }
    }
    @Data
    static class Messages{ 
        private Object obj;
        private String messageId;
        private String uuid;
    }
}

3、检测客户端服务端是否在线

一、服务端：

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.Delimiters;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.GlobalEventExecutor;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.TimeUnit;
/** * 启动服务端服务 * 需要开启一个新的线程来执行netty 服务端 或者客户端 * ThreadUtil.execute(() -> { * Server.initServer(); * Client.initClient(); * }); */
@Slf4j
public class Server { 
    /** * 装有所有客户端channel的组 */
    private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    static class ServerHeader extends SimpleChannelInboundHandler<String> { 
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, String msg) { 
            Channel channel = ctx.channel();
            //循环channel组，判断是不是其它客户端发送的消息
            channelGroup.stream().forEach(ch -> { 
                if (channel != ch) { 
                    ch.writeAndFlush(channel.remoteAddress() + "---发送的消息为：" + msg + "\n");
                } else { 
                    ch.writeAndFlush("【自己的消息】" + msg + "\n");
                }
            });
        }
        /** * 用户事件触发方法 判断事件 * * @param ctx 上下文对象 * @param evt 事件对象 */
        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) { 
            //判断该事件是否为超时事件
            if (evt instanceof IdleStateEvent) { 
                IdleStateEvent event = (IdleStateEvent) evt;
                String type;
                //判断具体事件 具体项目可以根据不同情况进行不同处理
                switch (event.state()) { 
                    case READER_IDLE:
                        type = "读空闲";
                        break;
                    case WRITER_IDLE:
                        type = "写空闲";
                        break;
                    default:
                        type = "读写空闲";
                }
                log.info("%s==超时事件==%s\n", ctx.channel().remoteAddress().toString(), type);
                ctx.channel().close();
            }
        }
        @Override
        public void handlerAdded(ChannelHandlerContext ctx) { 
            Channel channel = ctx.channel();
            //通知其它客户端
            channelGroup.writeAndFlush("【服务端：】" + channel.remoteAddress() + "加入\n");
            channelGroup.add(channel);
        }
        @Override
        public void channelActive(ChannelHandlerContext ctx) { 
            Channel channel = ctx.channel();
            log.info("【客户端：】" + channel.remoteAddress() + "上线\n");
        }
        @Override
        public void channelInactive(ChannelHandlerContext ctx) { 
            Channel channel = ctx.channel();
            log.info("【客户端：】" + channel.remoteAddress() + "下线\n");
        }
        @Override
        public void channelUnregistered(ChannelHandlerContext ctx) { 
            Channel channel = ctx.channel();
            channelGroup.writeAndFlush("【客户端：】" + channel.remoteAddress() + "离开\n");
            //此处不用从组中一处，掉线之后netty会自动的从组中移除
        }
    }
    static class ServerInit extends ChannelInitializer<SocketChannel> { 
        @Override
        protected void initChannel(SocketChannel socketChannel) { 
            ChannelPipeline pipeline = socketChannel.pipeline();
            //添加处理器，netty通过解决不同的情景来添加不同的处理器
            pipeline.addLast(new DelimiterBasedFrameDecoder(4096, Delimiters.lineDelimiter()))
                    .addLast(new StringDecoder())
                    .addLast(new StringEncoder())
                    .addLast(new IdleStateHandler(100,200,300, TimeUnit.SECONDS))
                    .addLast(new ServerHeader());
        }
    }
    /** * 初始化 */
    public static void initServer(){ 
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try { 
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup,workerGroup).channel(NioServerSocketChannel.class).childHandler(new ServerInit());
            // 端口可以自定义
            ChannelFuture channelFuture = serverBootstrap.bind(8899).sync();
            channelFuture.channel().closeFuture().sync();
        } catch (Exception e){ 
            log.error("初始化失败：{}",e);
        } finally { 
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

二、客户端：

import cn.hutool.cron.CronUtil;
import cn.hutool.cron.task.Task;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.Delimiters;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import lombok.extern.slf4j.Slf4j;
import java.io.BufferedReader;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
/** * 启动客户端服务 * 需要开启一个新的线程来执行netty 服务端 或者客户端 * ThreadUtil.execute(() -> { * Server.initServer(); * Client.initClient(); * }); */
@Slf4j
public class Client { 
    static class ClientHeader extends SimpleChannelInboundHandler<String> { 
        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, String msg) { 
            log.info("服务端消息：" + msg);
        }
    }
    static class ClientInit extends ChannelInitializer<SocketChannel> { 
        @Override
        protected void initChannel(SocketChannel socketChannel) { 
            ChannelPipeline pipeline = socketChannel.pipeline();
            //添加处理器，netty通过解决不同的情景来添加不同的处理器
            pipeline.addLast(new DelimiterBasedFrameDecoder(4096, Delimiters.lineDelimiter()))
                    .addLast(new StringDecoder())
                    .addLast(new StringEncoder())
                    .addLast(new ClientHeader());
        }
    }
    /** * 初始化客户端 */
    public static void initClient(){ 
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        try { 
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class).handler(new ClientInit());
            //获得channel对象
            Channel channel = bootstrap.connect("localhost", 8899).sync().channel();
            // 支持秒级别定时任务
            CronUtil.setMatchSecond(true);
            CronUtil.start();
            CronUtil.schedule("*/2 * * * * *", (Task) () -> { 
                String str = String.valueOf(System.currentTimeMillis());
                BufferedReader br = new BufferedReader(new InputStreamReader(new ByteArrayInputStream(str.getBytes())));
                try { 
                    channel.writeAndFlush(br.readLine() + "\r\n");
                } catch (IOException e) { 
                    log.error("通道已关闭：{}",e);
                }
            });
        } catch (Exception e) { 
            log.info("启动客户端失败：{}",e);
        } finally { 
            eventLoopGroup.shutdownGracefully();
        }
    }
}

三、springboot方式启动：

import cn.hutool.core.thread.ThreadUtil;
import com.imooc.stream.utils.Client;
import com.imooc.stream.utils.Server;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.boot.web.servlet.support.SpringBootServletInitializer;
@Slf4j
@SpringBootApplication
public class StreamApplication extends SpringBootServletInitializer implements CommandLineRunner { 
    public static void main(String[] args) { 
        SpringApplication.run(StreamApplication.class, args);
    }
    /** * 支持使用外置的tomcat启动 * @param application * @return */
    @Override
    protected SpringApplicationBuilder configure(SpringApplicationBuilder application) { 
        log.info("启动加载自定义的ServletInitializer");
        return application.sources(StreamApplication.class);
    }
    @Override
    public void run(String... args) { 
        log.info("启动成功, 同时启动netty服务");
        //需要开启一个新的线程来执行netty 服务端 或者客户端
        ThreadUtil.execute(() -> { 
            Server.initServer();
            Client.initClient();
        });
    }
}