RocketMQ5.0顺序消息设计实现

前言

顺序消息是 RocketMQ 提供的一种高级消息类型，支持消费者按照发送消息的先后顺序获取消息，从而实现业务场景中的顺序处理。
顺序消息的顺序关系通过消息组（MessageGroup）判定和识别，发送顺序消息时需要为每条消息设置归属的消息组，相同消息组的多条消息之间遵循先进先出的顺序关系，不同消息组、无消息组的消息之间不涉及顺序性。比如：一条订单从创建到完结整个生命周期内产生的消息，如果要保证消费的顺序性，则可以用订单号作为 MessageGroup。

RocketMQ 4.x 实现顺序消息相对容易，因为采用的是队列模型，一个队列只能被一个消费者消费，而队列本身是能保证先进先出的，此时只要保证消费者单线程串行消费即可。
到了 RocketMQ 5.0 时代，Pop 模式下因为采用的是消息模型，消费者可以消费所有队列的消息，顺序消息的实现也将变得更加复杂。

如何保证消息的顺序

顺序消息需要依赖生产者、Broker、消费者共同保证。

生产顺序性
首先是消息生产的顺序性，相同 MessageGroup 必须保证单一生产者、单线程同步发送。多生产者实例或者多线程并发发送消息，都无法保证消息是顺序到达 Broker 的，消息源头的顺序性都无法保证，后续流程的顺序就更是无从谈起了。

存储顺序性
消息发送到 Broker 必须按照到达顺序有序存储，这一点很容易实现。因为队列天生是先进先出的，但是一个 Topic 下可能会有多个队列，此时保证相同 MessageGroup 的消息被发送到同一个队列是重点，这个可以通过计算 MessageGroup 哈希值对队列数取模实现。

投递顺序性
消费者来拉取顺序消息时，Broker 得知道之前投递的消息是否全部被消费完了，如果还在消费中，则当前队列不能再继续投递了，消费者必须等待其它拉取到消息的消费者消费完毕后才能接着拉取后面的消息。

消费顺序性
消费者在拉取到消息后，必须保证单线程顺序消费，如果并发消费也是不能保证顺序的。

设计实现

生产端的顺序需要调用方自行保证，这个没啥好说的。
存储端的顺序，队列本身能保证先进先出，只要保证相同 MessageGroup 投递到同一个目标队列即可。Proxy 用一个叫 SendMessageQueueSelector 的组件对消息的 MessageGroup 计算一致性哈希后取模得到目标队列。

@Override
public AddressableMessageQueue select(ProxyContext ctx, MessageQueueView messageQueueView) {try {apache.rocketmq.v2.Message message = request.getMessages(0);String shardingKey = null;if (request.getMessagesCount() == 1) {// 分片键 也就是MessageGroupshardingKey = message.getSystemProperties().getMessageGroup();}AddressableMessageQueue targetMessageQueue;if (StringUtils.isNotEmpty(shardingKey)) {// 根据写队列数计算一致性哈希List<AddressableMessageQueue> writeQueues = messageQueueView.getWriteSelector().getQueues();int bucket = Hashing.consistentHash(shardingKey.hashCode(), writeQueues.size());targetMessageQueue = writeQueues.get(bucket);} else {targetMessageQueue = messageQueueView.getWriteSelector().selectOne(false);}return targetMessageQueue;} catch (Exception e) {return null;}
}

顺序消费

RocketMQ 5.0 消费者在启动时就会和 Proxy 建立 TCP 长连接，查询订阅的 Topic 路由数据TopicRouteData。紧接着调用 telemetry 接口发送 SETTINGS 命令同步设置，要同步哪些设置呢？
RocketMQ 5.0 的客户端 SDK 要做轻量化，客户端最好啥也不知道，一些策略和配置最好靠服务端下发，同步设置就是干这个的。

请求体表明了消费者所属的消费组，以及消息订阅配置：

client_type: PUSH_CONSUMER
access_point {scheme: IPv4addresses {host: "127.0.0.1"port: 8081}
}
request_timeout {seconds: 3
}
subscription {group {name: "G_fifo"}subscriptions {topic {name: "fifo"}expression {type: TAGexpression: "*"}}
}
user_agent {language: JAVAversion: "5.0.4"platform: "Mac OS X 10.16"hostname: "localhost-5.local"
}

Proxy 返回的设置信息包含：消息消费失败的重试策略、消费者是否要顺序消费、单次最大消息拉取数量、以及无消息时的长轮询挂起时间。

client_type: PUSH_CONSUMER
access_point {scheme: IPv4addresses {host: "127.0.0.1"port: 8081}
}
backoff_policy {max_attempts: 17customized_backoff {next {seconds: 1}next {seconds: 5}next {seconds: 10}next {seconds: 30}next {seconds: 60}next {seconds: 120}next {seconds: 180}next {seconds: 240}next {seconds: 300}next {seconds: 360}next {seconds: 420}next {seconds: 480}next {seconds: 540}next {seconds: 600}next {seconds: 1200}next {seconds: 1800}next {seconds: 3600}next {seconds: 7200}}
}
request_timeout {seconds: 3
}
subscription {group {name: "G_fifo"}subscriptions {topic {name: "fifo"}expression {type: TAGexpression: "*"}}fifo: truereceive_batch_size: 32long_polling_timeout {seconds: 20}
}
user_agent {language: JAVAversion: "5.0.4"platform: "Mac OS X 10.16"hostname: "localhost-5.local"
}
metric {
}

对于顺序消息来说，最重要的配置项就是fifo: true，它决定了消费者是多线程并发消费还是单线程串行消费，消费者会根据配置创建对应的 ConsumeService。顾名思义，FifoConsumeService 是用来消费顺序消息的，StandardConsumeService 用来消费普通消息。

private ConsumeService createConsumeService() {final ScheduledExecutorService scheduler = this.getClientManager().getScheduler();if (pushSubscriptionSettings.isFifo()) {return new FifoConsumeService(clientId, messageListener, consumptionExecutor, this, scheduler);}return new StandardConsumeService(clientId, messageListener, consumptionExecutor, this, scheduler);
}

FifoConsumeService 会按照顺序消费拉取到的消息，而且会等待上一个消息消费完毕才会去消费下一个。

@Override
public void consume(ProcessQueue pq, List<MessageViewImpl> messageViews) {// 基于迭代器消费consumeIteratively(pq, messageViews.iterator());
}public void consumeIteratively(ProcessQueue pq, Iterator<MessageViewImpl> iterator) {if (!iterator.hasNext()) {return;}final MessageViewImpl messageView = iterator.next();if (messageView.isCorrupted()) {// 消息损坏consumeIteratively(pq, iterator);return;}// 触发MessageListener消费消息final ListenableFuture<ConsumeResult> future0 = consume(messageView);// 处理消费结果ListenableFuture<Void> future = Futures.transformAsync(future0, result -> pq.eraseFifoMessage(messageView,result), MoreExecutors.directExecutor());// 等待消息消费完毕再递归消费下一个消息future.addListener(() -> consumeIteratively(pq, iterator), MoreExecutors.directExecutor());
}

对于顺序消息来说，消费失败是个麻烦事儿。因为要保证消息的顺序，上一个消息没消费成功，下一个消息就无法被消费，容易导致消息堆积。RocketMQ 的策略是重试几次，还是不行就发到死信队列，方法是ProcessQueueImpl#eraseFifoMessage

@Override
public ListenableFuture<Void> eraseFifoMessage(MessageViewImpl messageView, ConsumeResult consumeResult) {statsConsumptionResult(consumeResult);final RetryPolicy retryPolicy = consumer.getRetryPolicy();// 最大重试次数final int maxAttempts = retryPolicy.getMaxAttempts();int attempt = messageView.getDeliveryAttempt();final MessageId messageId = messageView.getMessageId();final ConsumeService service = consumer.getConsumeService();final ClientId clientId = consumer.getClientId();// 失败且没超过最大重试次数if (ConsumeResult.FAILURE.equals(consumeResult) && attempt < maxAttempts) {// 下一个延迟时间final Duration nextAttemptDelay = retryPolicy.getNextAttemptDelay(attempt);attempt = messageView.incrementAndGetDeliveryAttempt();log.debug("Prepare to redeliver the fifo message because of the consumption failure, maxAttempt={}," +" attempt={}, mq={}, messageId={}, nextAttemptDelay={}, clientId={}", maxAttempts, attempt, mq,messageId, nextAttemptDelay, clientId);// 丢到线程池定时调度执行final ListenableFuture<ConsumeResult> future = service.consume(messageView, nextAttemptDelay);return Futures.transformAsync(future, result -> eraseFifoMessage(messageView, result),MoreExecutors.directExecutor());}boolean ok = ConsumeResult.SUCCESS.equals(consumeResult);// 超过重试次数还是失败 发到死信队列ListenableFuture<Void> future = ok ? ackMessage(messageView) : forwardToDeadLetterQueue(messageView);future.addListener(() -> evictCache(messageView), consumer.getConsumptionExecutor());return future;
}

至此，消费端对于顺序消息的处理就结束了。核心是如果消费组配置的是顺序投递，消费者在拉取到消息后会单线程同步消费消息。

顺序投递

消费者的顺序性还是比较容易保证的，整个链路里最复杂的必须是 Broker 投递的顺序性，因为 Broker 得记录队列里上一批拉取到的消息是否全部消费完，根据此来判断要不要继续投递后面的消息。
Broker 引入一个新组件 ConsumerOrderInfoManager，来管理消费者顺序消息的消费情况。它继承了 ConfigManager，所以支持数据的持久化。
image.png
它内部使用一个双层嵌套 Map 来记录消费组对于某个队列的顺序消息消费情况，所谓的数据持久化就是把这个 Map 序列化成 JSON 后落地到磁盘。

private ConcurrentHashMap<String/* topic@group*/, ConcurrentHashMap<Integer/*queueId*/, OrderInfo>> table =new ConcurrentHashMap<>(128);

落盘的文件路径是{storeHome}/config/consumerOrderInfo.json，内容大概长这样：

{"table":{"fifo@G_fifo":{0:{"cm":1,"i":60000,"l":1703644544701,"o":[460],"oc":{},"popTime":1703644544701}}}
}

核心是 OrderInfo 类，它记录了消费者针对某个队列拉取到的最新一批顺序消息的消费情况。offsetList 记录了消息的偏移量，可以根据此来定位消息；commitOffsetBit 记录了各消息的消费情况，它是一个位图，消息提交以后会把对应的比特位设为1。

public static class OrderInfo {// 各消息的偏移量(增量编码)private List<Long> offsetList;// 消耗次数private int consumedCount;// 最近一次消费的时间戳 其实是拉取时间private long lastConsumeTimestamp;// 消息提交位图private long commitOffsetBit;
}

消费者在拉取消息时，Broker 会给投递的这一批顺序消息记录一个 OrderInfo

private long popMsgFromQueue() {......if (isOrder) {// 顺序消息 给拉取到的这一批消息记录OrderInfoint count = brokerController.getConsumerOrderInfoManager().update(topic,requestHeader.getConsumerGroup(),queueId, getMessageTmpResult.getMessageQueueOffset());this.brokerController.getConsumerOffsetManager().commitOffset(channel.remoteAddress().toString(),requestHeader.getConsumerGroup(), topic, queueId, offset);ExtraInfoUtil.buildOrderCountInfo(orderCountInfo, isRetry, queueId, count);} else {// 普通消息 追加CheckPointappendCheckPoint(requestHeader, topic, reviveQid, queueId, offset, getMessageTmpResult, popTime, this.brokerController.getBrokerConfig().getBrokerName());}......
}

方法是ConsumerOrderInfoManager#update，主要是构建一个 OrderInfo 对象存入 Map

public int update(String topic, String group, int queueId, List<Long> msgOffsetList) {String key = topic + TOPIC_GROUP_SEPARATOR + group;ConcurrentHashMap<Integer/*queueId*/, OrderInfo> qs = table.get(key);if (qs == null) {qs = new ConcurrentHashMap<>(16);ConcurrentHashMap<Integer/*queueId*/, OrderInfo> old = table.putIfAbsent(key, qs);if (old != null) {qs = old;}}OrderInfo orderInfo = qs.get(queueId);// 转增量编码List<Long> simple = OrderInfo.simpleO(msgOffsetList);if (orderInfo != null && simple.get(0).equals(orderInfo.getOffsetList().get(0))) {if (simple.equals(orderInfo.getOffsetList())) {orderInfo.setConsumedCount(orderInfo.getConsumedCount() + 1);} else {// reset, because msgs are changed.orderInfo.setConsumedCount(0);}orderInfo.setLastConsumeTimestamp(System.currentTimeMillis());orderInfo.setOffsetList(simple);orderInfo.setCommitOffsetBit(0);} else {// 构建新的OrderInfo覆盖掉上一批orderInfo = new OrderInfo();orderInfo.setOffsetList(simple);orderInfo.setLastConsumeTimestamp(System.currentTimeMillis());orderInfo.setConsumedCount(0);orderInfo.setCommitOffsetBit(0);qs.put(queueId, orderInfo);}return orderInfo.getConsumedCount();
}

假设此时又有其它消费者来拉取同一队列的消息，Broker 会先定位到对应的 OrderInfo，再判断是否要继续投递后面的消息：

private long popMsgFromQueue() {......if (isOrder && brokerController.getConsumerOrderInfoManager().checkBlock(topic,requestHeader.getConsumerGroup(), queueId, requestHeader.getInvisibleTime())) {// 之前拉取的一批消息还没全部commit，不能拉取新消息return this.brokerController.getMessageStore().getMaxOffsetInQueue(topic, queueId) - offset + restNum;}......
}

方法是ConsumerOrderInfoManager#checkBlock，只有当下面两个条件都满足，Broker 才会拒绝投递：

  • 上一批消息的拉取时间还没超过消息的不可见时间(60s)
  • 上一批消息还没全部提交
public boolean checkBlock(String topic, String group, int queueId, long invisibleTime) {String key = topic + TOPIC_GROUP_SEPARATOR + group;ConcurrentHashMap<Integer/*queueId*/, OrderInfo> qs = table.get(key);if (qs == null) {qs = new ConcurrentHashMap<>(16);ConcurrentHashMap<Integer/*queueId*/, OrderInfo> old = table.putIfAbsent(key, qs);if (old != null) {qs = old;}}OrderInfo orderInfo = qs.get(queueId);if (orderInfo == null) {// 当前队列还没拉取过，可以直接拉return false;}// 距离最后一次消费时间是否小于不可见时间60sboolean isBlock = System.currentTimeMillis() - orderInfo.getLastConsumeTimestamp() < invisibleTime;/*** 没超过不可见时间，则必须等这一批消息全部commit才能继续拉取*/return isBlock && !orderInfo.isDone();
}

判断消息是否全部提交的方法是OrderInfo#isDone，其实就是判断 commitOffsetBit 位图对应的位是否全部为1：

public boolean isDone() {if (offsetList == null || offsetList.isEmpty()) {return true;}int num = offsetList.size();for (byte i = 0; i < num; i++) {if ((commitOffsetBit & (1L << i)) == 0) {return false;}}return true;
}

消息投递后，消费者会按照顺序串行消费并上报消费结果，即 ack 消息。Broker 在处理消息的 ack 请求时会判断 ack 的是不是顺序消息，如果是就会更新 OrderInfo 位图。然后再判断 OrderInfo 里的这一批消息是否全部提交，如果是就提交消费位点，同时通知其它被挂起的请求拉取消息。

private RemotingCommand processRequest(){......if (rqId == KeyBuilder.POP_ORDER_REVIVE_QUEUE) {// 顺序消息String lockKey = requestHeader.getTopic() + PopAckConstants.SPLIT+ requestHeader.getConsumerGroup() + PopAckConstants.SPLIT + requestHeader.getQueueId();long oldOffset = this.brokerController.getConsumerOffsetManager().queryOffset(requestHeader.getConsumerGroup(),requestHeader.getTopic(), requestHeader.getQueueId());if (requestHeader.getOffset() < oldOffset) {return response;}// 加锁while (!this.brokerController.getPopMessageProcessor().getQueueLockManager().tryLock(lockKey)) {}try {oldOffset = this.brokerController.getConsumerOffsetManager().queryOffset(requestHeader.getConsumerGroup(),requestHeader.getTopic(), requestHeader.getQueueId());if (requestHeader.getOffset() < oldOffset) {return response;}// 更新位图long nextOffset = brokerController.getConsumerOrderInfoManager().commitAndNext(requestHeader.getTopic(), requestHeader.getConsumerGroup(),requestHeader.getQueueId(), requestHeader.getOffset());if (nextOffset > -1) {// 这一批顺序消息全部消费掉了，提交消费位点this.brokerController.getConsumerOffsetManager().commitOffset(channel.remoteAddress().toString(),requestHeader.getConsumerGroup(), requestHeader.getTopic(),requestHeader.getQueueId(),nextOffset);// 通知其它被挂起的请求开始拉取消息this.brokerController.getPopMessageProcessor().notifyMessageArriving(requestHeader.getTopic(), requestHeader.getConsumerGroup(),requestHeader.getQueueId());} else if (nextOffset == -1) {String errorInfo = String.format("offset is illegal, key:%s, old:%d, commit:%d, next:%d, %s",lockKey, oldOffset, requestHeader.getOffset(), nextOffset, channel.remoteAddress());POP_LOGGER.warn(errorInfo);response.setCode(ResponseCode.MESSAGE_ILLEGAL);response.setRemark(errorInfo);return response;}} finally {this.brokerController.getPopMessageProcessor().getQueueLockManager().unLock(lockKey);}return response;}......
}

更新位图的方法是ConsumerOrderInfoManager#commitAndNext，它会把 commitOffsetBit 对应的比特位设为1，然后返回值代表消息是否全被消费掉了，通知外层要提交消费位点。-2 代表还没消费完、大于等于0表示需要提交消费位点。

public long commitAndNext(String topic, String group, int queueId, long offset) {String key = topic + TOPIC_GROUP_SEPARATOR + group;ConcurrentHashMap<Integer/*queueId*/, OrderInfo> qs = table.get(key);if (qs == null) {return offset + 1;}OrderInfo orderInfo = qs.get(queueId);if (orderInfo == null) {log.warn("OrderInfo is null, {}, {}, {}", key, offset, orderInfo);return offset + 1;}List<Long> offsetList = orderInfo.getOffsetList();if (offsetList == null || offsetList.isEmpty()) {log.warn("OrderInfo is empty, {}, {}, {}", key, offset, orderInfo);return -1;}Long first = offsetList.get(0);int i = 0, size = offsetList.size();for (; i < size; i++) {long temp;if (i == 0) {temp = first;} else {temp = first + offsetList.get(i);}if (offset == temp) {break;}}if (i >= size) {log.warn("OrderInfo not found commit offset, {}, {}, {}", key, offset, orderInfo);return -1;}// 更新Commit位图 对应位设为1orderInfo.setCommitOffsetBit(orderInfo.getCommitOffsetBit() | (1L << i));if (orderInfo.isDone()) {// 这一批消息全部Commit了if (size == 1) {return offsetList.get(0) + 1;} else {return offsetList.get(size - 1) + first + 1;}}// 无需commitreturn -2;
}

尾巴

RocketMQ 顺序消息需要多端共同保证，包括：生产端顺序性、存储端顺序性、投递端顺序性、消费端顺序性。5.0 和 4.x 最大的区别就是，Pop 模式下的消息模型允许消费者消费所有队列，Broker 投递的顺序性是实现难点。RocketMQ 给出的解决方案是用一个嵌套 Map 维护 OrderInfo，用来管理消费组针对某个队列的消费情况。Broker 在投递消息前会针对这一批消息构建一个 OrderInfo 对象存储下来，在收到消费者发送的 ack 请求时更新对应的位图。下一个消费者来拉取消息时，Broker 会判断对应的 OrderInfo 里的消息是否全部提交，如果还有消息没提交，是不会投递后面的消息的，以此来保证消息投递的顺序性。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/317724.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

五步解决Ubuntu界面太小的问题

五步解决Ubuntu界面太小的问题

名人说&#xff1a;莫听穿林打叶声&#xff0c;何妨吟啸且徐行。—— 苏轼《定风波莫听穿林打叶声》 Code_流苏(CSDN)&#xff08;一个喜欢古诗词和编程的Coder&#xff09; 对于20版本及以上的unbuntu我们可以通过安装open-vm-tools来解决界面大小的问题&#xff0c;具体步骤如…
阅读更多...
Leetcode 62 不同路径

Leetcode 62 不同路径

题意理解&#xff1a; 一个机器人位于一个 m x n 网格的左上角 &#xff08;起始点在下图中标记为 “Start” &#xff09;。 机器人每次只能向下或者向右移动一步。机器人试图达到网格的右下角&#xff08;在下图中标记为 “Finish” &#xff09; 要求&#xff1a;机器人只能…
阅读更多...
定时器PWM控制RGB彩灯案例

定时器PWM控制RGB彩灯案例

1.脉冲宽度调制PWM PWM&#xff08;Pulse Width Modulation&#xff09;简称脉宽调制&#xff0c;是利用微处理器的数字输出来对模拟电路进行控制的一种非常有效的技术&#xff0c;广泛应用在测量、通信、工控等方面。   PWM的一个优点是从处理器到​​ ​被控系统​​​信号…
阅读更多...
vue assets动态引入图片解决打包后图片路径错误不显示的问题

vue assets动态引入图片解决打包后图片路径错误不显示的问题

通用&#xff1a;适用于处理单个链接的资源文件 import homeIcon from /assets/images/home/icon.png <img :src"homeIcon" /> webpack 构建的项目 这里我们先假设&#xff1a; 静态文件目录&#xff1a;src/assets/images/ 我们的目标静态文件在 src/assets…
阅读更多...
解决Redis序列化乱码问题

解决Redis序列化乱码问题

如果我们使用原生的JDK序列化&#xff0c;那么当我们将数据存储到Redis中就会出现乱码的情况 为了解决这个问题我们需要重写RedisTemplate从而解决序列化乱码问题 首先在Maven中引入相应的依赖 <dependency> <groupId>com.fasterxml.jackson.core</group…
阅读更多...
【每日一题】466. 统计重复个数-2024.1.2

【每日一题】466. 统计重复个数-2024.1.2

题目&#xff1a; 466. 统计重复个数 定义 str [s, n] 表示 str 由 n 个字符串 s 连接构成。 例如&#xff0c;str ["abc", 3] "abcabcabc" 。 如果可以从 s2 中删除某些字符使其变为 s1&#xff0c;则称字符串 s1 可以从字符串 s2 获得。 例如&…
阅读更多...
如何使用凹凸贴图和位移贴图制作逼真的模型

如何使用凹凸贴图和位移贴图制作逼真的模型

在线工具推荐&#xff1a; 3D数字孪生场景编辑器 - GLTF/GLB材质纹理编辑器 - 3D模型在线转换 - Three.js AI自动纹理开发包 - YOLO 虚幻合成数据生成器 - 三维模型预览图生成器 - 3D模型语义搜索引擎 本教程将解释如何应用这些效应背后的理论。在以后的教程中&#xff0…
阅读更多...
【完整代码】网上书店信息管理系统--基于Mysql数据库与java

【完整代码】网上书店信息管理系统--基于Mysql数据库与java

网上书店信息管理系统 一、需求分析&#xff08;一&#xff09;设计系统的意义以及用途&#xff08;二&#xff09;实现的功能1.用户模块&#xff1a;1、全部图书浏览2、图书搜索3、购物车管理和订单查看4、修改密码 2.书店管理员模块1、图书类别管理2、图书管理3、全部订单查看…
阅读更多...
无辅源电压继电器 RWY-D2/3 180-440VAC 导轨安装 josef约瑟

无辅源电压继电器 RWY-D2/3 180-440VAC 导轨安装 josef约瑟

RWY-D1型电压继电器&#xff1b; RWY-D2型电压继电器&#xff1b; 一、 概述 RWY-D系列电压继电器&#xff08;以下简称本继电器&#xff09;用于发电机、变压器和输电线的电器保护装置中&#xff0c;作为过电压保护或低电压闭锁的启动原件。本继电器为集成电路静态型继电器…
阅读更多...
《数字图像处理-OpenCV/Python》连载：空间滤波之高斯滤波器

《数字图像处理-OpenCV/Python》连载：空间滤波之高斯滤波器

《数字图像处理-OpenCV/Python》连载&#xff1a;空间滤波之高斯滤波器 本书京东 优惠购书链接 https://item.jd.com/14098452.html 本书CSDN 独家连载专栏 https://blog.csdn.net/youcans/category_12418787.html 第 10 章 图像卷积与空间滤波 图像滤波是指在尽可能保留图像细…
阅读更多...
UOS下通过SSH隧道访问云端内网windows桌面

UOS下通过SSH隧道访问云端内网windows桌面

1 用户痛点 随着时代的发展&#xff0c;众多企业的服务器慢慢走向云端。大量云端服务器节省企业成本的同时&#xff0c;也带来了安全性问题。例如&#xff1a;管理云端的服务器&#xff0c;特别是windows桌面服务器&#xff0c;往往需要给这个服务器分配一个公网IP地址&#x…
阅读更多...
面向对象综合训练综合练习(文字版格斗游戏,对象数组,复杂的对象数组操作)

面向对象综合训练综合练习(文字版格斗游戏,对象数组,复杂的对象数组操作)

文章目录 练习一&#xff1a;文字版格斗游戏练习二&#xff1a;文字版格斗游戏进阶练习三&#xff1a;对象数组&#xff08;商品&#xff09;练习四&#xff1a;对象数组&#xff08;汽车&#xff09;练习五&#xff1a;对象数组&#xff08;手机&#xff09;练习六&#xff1a…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023