Mqtt设备模拟器（RPC）

package com.yeexun.xiot.simulator.srv.mqttSimulator; import com.alibaba.fastjson.JSON; import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; import com.google.common.collect.Maps; import com.yeexun.xiot.base.dto.R; import com.yeexun.xiot.simulator.srv.message.MessageHandler; import com.yeexun.xiot.simulator.srv.web.domain.dto.StartDto; import com.yeexun.xiot.simulator.srv.web.domain.entity.MQTTAuth; import com.yeexun.xiot.simulator.srv.web.domain.entity.XiotSimMessage; import com.yeexun.xiot.simulator.srv.web.mapper.XiotSimMessageMapper; import com.yeexun.xiot.simulator.srv.web.service.SimulatorUserService; import io.netty.buffer.Unpooled; import io.netty.channel.EventLoopGroup; import io.netty.channel.epoll.Epoll; import io.netty.channel.epoll.EpollEventLoopGroup; import io.netty.channel.epoll.EpollSocketChannel; import io.netty.channel.kqueue.KQueue; import io.netty.channel.kqueue.KQueueEventLoopGroup; import io.netty.channel.kqueue.KQueueSocketChannel; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.handler.codec.mqtt.MqttConnectReturnCode; import io.netty.handler.codec.mqtt.MqttVersion; import io.netty.util.concurrent.Future; import lombok.AllArgsConstructor; import lombok.Getter; import lombok.extern.slf4j.Slf4j; import org.hswebframework.expands.script.engine.DynamicScriptEngine; import org.hswebframework.expands.script.engine.DynamicScriptEngineFactory; import org.jetlinks.mqtt.client.MqttClient; import org.jetlinks.mqtt.client.MqttClientCallback; import org.jetlinks.mqtt.client.MqttClientConfig; import org.jetlinks.mqtt.client.MqttConnectResult; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; import org.springframework.context.annotation.Scope; import org.springframework.stereotype.Service; import javax.annotation.Resource; import java.math.BigDecimal; import java.net.InetSocketAddress; import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; import java.util.LinkedHashMap; import java.util.Map; import java.util.Random; import java.util.Set; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.LongAdder; import java.util.function.BiConsumer; import java.util.function.Consumer; /** * <p> * 模拟器实现类 * </p> * * @author wangtao * @since 2021-12-02 */ @Service("MqttSimulator") @Scope("prototype") @Slf4j public class MqttSimulator implements SimulatorUserService { private BlockingQueue<Runnable> createMqttJob; private ScheduledExecutorService executorService; @Autowired private StartDto startDto; public MqttSimulator(StartDto startDto) { this.startDto = startDto; } private int threadSize = Runtime.getRuntime().availableProcessors() * 2; Map<String, ClientSession> clientMap = new ConcurrentHashMap<>(); BiConsumer<Integer, ClientSession> eventDataSuppliers; BiConsumer<Integer, ClientSession> propertyDataSuppliers; Map<String, MessageHandler> messageHandlerMap = new HashMap<>(); private Map<String, AtomicInteger> portCounter = new ConcurrentHashMap<>(); private Consumer<ClientSession> onConnect; // ArrayList<ScheduledExecutorService> scheduledExecutorServices=new ArrayList<>(); transient EventLoopGroup eventLoopGroup; transient Class channelClass; private static Map<String, ScheduledFuture> futureMap = new ConcurrentHashMap<>(); private static Map<String, ScheduledExecutorService> serviceMap = new ConcurrentHashMap<>(); private static Map<String, Map<String, MqttClient>> taskMap = new ConcurrentHashMap<>(); private static Map<String, Map<String, ClientSession>> clientSessionMap = new ConcurrentHashMap<>(); public boolean addTask(ScheduledTask scheduledTask, ScheduledExecutorService executorService) { if (futureMap.get(scheduledTask.getTaskId()) != null) { return false; } ScheduledFuture<?> scheduledFuture = executorService.scheduleAtFixedRate(scheduledTask.getRunnable(), scheduledTask.getInitialDelay(), scheduledTask.getPeriod(), TimeUnit.MILLISECONDS); futureMap.put(scheduledTask.getTaskId(), scheduledFuture); return true; } public boolean remove(XiotSimMessage simulator) { if (futureMap.get(simulator.getId()) == null) { return false; } ScheduledFuture scheduledFuture = futureMap.get(simulator.getId()); scheduledFuture.cancel(false); futureMap.remove(simulator.getId()); ScheduledExecutorService scheduledExecutorService = serviceMap.get(simulator.getId()); scheduledExecutorService.shutdown(); serviceMap.remove(simulator.getId()); Map<String, ClientSession> stringClientSessionMap = clientSessionMap.get(simulator.getId()); for (int i = 0; i < (simulator.getStart() + simulator.getTotal()); i++) { ClientSession clientSession = stringClientSessionMap.get(simulator.getClientId() + i); // System.out.println(clientSession.getClient().getClientConfig().getClientId()); clientSession.getClient().disconnect(); } taskMap.remove(simulator.getId()); return true; } public void shutdown() { executorService.shutdown(); } public void runRate(Runnable runnable, long time) { executorService.scheduleAtFixedRate(runnable, 2000, time, TimeUnit.MILLISECONDS); } public void init(StartDto startDto) { executorService = Executors.newScheduledThreadPool(startDto.getBatchSize()); serviceMap.put(startDto.getId(), executorService); createMqttJob = new LinkedBlockingQueue<>(startDto.getBatchSize()); if (KQueue.isAvailable()) { eventLoopGroup = new KQueueEventLoopGroup(threadSize); channelClass = KQueueSocketChannel.class; } else if (Epoll.isAvailable()) { eventLoopGroup = new EpollEventLoopGroup(threadSize); channelClass = EpollSocketChannel.class; } else { eventLoopGroup = new NioEventLoopGroup(threadSize); channelClass = NioSocketChannel.class; } } @Override public R stop(XiotSimMessage simulator) { boolean remove = this.remove(simulator); if (!remove) { return new R<>(R.FAIL, "停止失败", null); } return new R<>(R.SUCCESS, "停止成功", null); } @Override public R start() throws Exception { String[] binds = startDto.getBinds().split(","); System.out.println("************" + binds); init(startDto); //String s = new String(Files.readAllBytes(Paths.get(scriptFile))); String s = startDto.getContent(); DynamicScriptEngine js = DynamicScriptEngineFactory.getEngine("js"); js.compile("handler", s); HashMap<String, Object> context = Maps.newHashMap(); context.put("simulator", this); context.put("logger", LoggerFactory.getLogger("message.handler")); js.execute("handler", context).getIfSuccess(); int end = startDto.getStart() + startDto.getTotal(); AtomicInteger started = new AtomicInteger();//用于诸如原子递增计数器之类的应用程序 int queueSizePeekSize = Math.