发表于

这是我参与11月更文挑战的第22天，活动详情查看：2021最后一次更文挑战

题目

给你一个整数数组 nums ，设计算法来打乱一个没有重复元素的数组。

实现 Solution class:

  • Solution(int[] nums) 使用整数数组 nums 初始化对象
  • int[] reset() 重设数组到它的初始状态并返回
  • int[] shuffle() 返回数组随机打乱后的结果

示例

1
2
3
4
5
6
7
8
9
10
11
css复制代码输入
["Solution", "shuffle", "reset", "shuffle"]
[[[1, 2, 3]], [], [], []]
输出
[null, [3, 1, 2], [1, 2, 3], [1, 3, 2]]

解释
Solution solution = new Solution([1, 2, 3]);
solution.shuffle();    // 打乱数组 [1,2,3] 并返回结果。任何 [1,2,3]的排列返回的概率应该相同。例如，返回 [3, 1, 2]
solution.reset();      // 重设数组到它的初始状态 [1, 2, 3] 。返回 [1, 2, 3]
solution.shuffle();    // 随机返回数组 [1, 2, 3] 打乱后的结果。例如，返回 [1, 3, 2]

提示

  • 1 <= nums.length <= 200
  • -10^6 <= nums[i] <= 10^6
  • nums 中的所有元素都是 唯一的
  • 最多可以调用 5 * 10^4 次 reset 和 shuffle

解题思路

Random随机数

题目中要求我们需要实现两个方法，一个是返回打乱顺序后的数组，一个是返回原数组。对于返回原数组的，我们可以采用空间换时间的方式，定义多一个数组orig用来保存原数组，当调用reset()方法当时候直接将结果返回即可。

至于打乱顺序的方法，我们可以直接使用现成的API Random 来实现。通过随机交换数组中元素位置，实现乱序。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
java复制代码class Solution {
    private static final Random random = new Random();
    private int[] orig;
    private int[] nums;

    public Solution(int[] nums) {
        this.orig = nums;
        this.nums = new int[nums.length];
        int index = 0;
        for(int num : nums){
            this.nums[index++] = num;
        }
    }
    
    public int[] reset() {
        // 直接返回原数组
        return orig;
    }
    
    public int[] shuffle() {
        int n = nums.length;
        for(int i = 0; i < n; ++i){
            // 随机取得一位元素进行交换
            int j = random.nextInt(n);
            nums[i] = nums[i] + nums[j] - (nums[j] = nums[i]);
        }
        return nums;
    }
}

复杂度分析

  • 时间复杂度：O(N)O(N)O(N)
  • 空间复杂度：O(N)O(N)O(N)

最后

文章有写的不好的地方，请大佬们不吝赐教，错误是最能让人成长的，愿我与大佬间的距离逐渐缩短！

如果觉得文章对你有帮助，请 点赞、收藏、关注、评论 一键四连支持，你的支持就是我创作最大的动力！！！

题目出处: leetcode-cn.com/problems/sh…

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

这是我参与11月更文挑战的第22天，活动详情查看：2021最后一次更文挑战

RocketMQ的前提回顾

RocketMQ是一款分布式、队列模型的消息中间件，具有以下特点：

  1. 能够保证严格的消息顺序
  2. 提供丰富的消息拉取模式
  3. 高效的订阅者水平扩展能力
  4. 实时的消息订阅机制
  5. 亿级消息堆积能力

为什么使用RocketMQ

  1. 强调集群无单点，可扩展，任意一点高可用、水平可扩展
  2. 海量消息堆积能力，消息堆积后写入低延迟
  3. 支持上万个队列
  4. 消息失败重试机制
  5. 消息可查询
  6. 开源社区活跃
  7. 成熟度已经经过淘宝双十一的考验

RocketMQ的发展变化

RocketMQ开源是使用文件作为持久化工具，阿里内部未开源的性能会更高，使用oceanBase作为持久化工具。
在RocketMQ1.x和2.x使用zookeeper管理集群，3.x开始使用nameserver代替zk，更轻量级，此外RocketMQ的客户端拥有两种的操作方式：DefaultMQPushConsumer和DefaultMQPullConsumer。

DefaultMQPushConsumer的Maven配置

1
2
3
4
5
xml复制代码<dependency>
   <groupId>org.apache.rocketmq</groupId>
   <artifactId>rocketmq-client</artifactId>
   <version>4.3.0</version>
</dependency>

DefaultMQPushConsumer使用示例

  1. CONSUME_FROM_LAST_OFFSET：第一次启动从队列最后位置消费，后续再启动接着上次消费的进度开始消费
  2. CONSUME_FROM_FIRST_OFFSET：第一次启动从队列初始位置消费，后续再启动接着上次消费的进度开始消费
  3. CONSUME_FROM_TIMESTAMP：第一次启动从指定时间点位置消费，后续再启动接着上次消费的进度开始消费

以上所说的第一次启动是指从来没有消费过的消费者，如果该消费者消费过，那么会在broker端记录该消费者的消费位置，如果该消费者挂了再启动，那么自动从上次消费的进度开始

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
java复制代码public class MQPushConsumer {
    public static void main(String[] args) throws MQClientException {
        String groupName = "rocketMqGroup1";
        // 用于把多个Consumer组织到一起，提高并发处理能力
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
        // 设置nameServer地址，多个以;分隔
        consumer.setNamesrvAddr("name-serverl-ip:9876;name-server2-ip:9876"); consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setMessageModel(MessageModel.BROADCASTING);
        // 订阅topic，可以对指定消息进行过滤，例如："TopicTest","tagl||tag2||tag3",*或null表示topic所有消息
        consumer.subscribe("order-topic", "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> mgs,
                    ConsumeConcurrentlyContext consumeconcurrentlycontext) {
                System.out.println(Thread.currentThread().getName()+"Receive New Messages:"+mgs);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
    }
}
  • CLUSTERING：默认模式，同一个ConsumerGroup(groupName相同)每个consumer只消费所订阅消息的一部分内容，同一个ConsumerGroup里所有的Consumer消息加起来才是所
  • 订阅topic整体，从而达到负载均衡的目的
  • BROADCASTING：同一个ConsumerGroup每个consumer都消费到所订阅topic所有消息，也就是一个消费会被多次分发，被多个consumer消费。

ConsumeConcurrentlyStatus.RECONSUME_LATER boker会根据设置的messageDelayLevel发起重试，默认16次。

DefaultMQPushConsumerImpl中各个对象的主要功能如下：

RebalancePushImpl：主要负责决定，当前的consumer应该从哪些Queue中消费消息；

  • 1）PullAPIWrapper：长连接，负责从broker处拉取消息，然后利用ConsumeMessageService回调用户的Listener执行消息消费逻辑；
  • 2）ConsumeMessageService：实现所谓的”Push-被动”消费机制；从Broker拉取的消息后，封装成ConsumeRequest提交给ConsumeMessageSerivce，此service负责回调用户的Listener消费消息；
  • 3）OffsetStore：维护当前consumer的消费记录（offset）；有两种实现，Local和Rmote，Local存储在本地磁盘上，适用于BROADCASTING广播消费模式；而Remote则将消费进度存储在Broker上，适用于CLUSTERING集群消费模式；
  • 4）MQClientFactory：负责管理client（consumer、producer），并提供多中功能接口供各个Service（Rebalance、PullMessage等）调用；大部分逻辑均在这个类中完成；

consumer.registerMessageListener执行过程：

1
2
3
4
5
6
7
8
java复制代码/**
     * Register a callback to execute on message arrival for concurrent consuming.
     * @param messageListener message handling callback.
     */
    @Override
    public void registerMessageListener(MessageListenerConcurrently messageListener) {
        this.messageListener = messageListener; this.defaultMQPushConsumerImpl.registerMessageListener(messageListener);
    }

通过源码可以看出主要实现过程在DefaultMQPushConsumerImpl类中consumer.start后调用DefaultMQPushConsumerImpl的同步start方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
java复制代码public synchronized void start() throws MQClientException {
        switch (this.serviceState) {
            case CREATE_JUST:
                log.info("the consumer [{}] start beginning. messageModel={}, isUnitMode={}", this.defaultMQPushConsumer.getConsumerGroup(),
                    this.defaultMQPushConsumer.getMessageModel(), this.defaultMQPushConsumer.isUnitMode());
                this.serviceState = ServiceState.START_FAILED;
                this.checkConfig();
                this.copySubscription();
                if (this.defaultMQPushConsumer.getMessageModel() == MessageModel.CLUSTERING) {
                    this.defaultMQPushConsumer.changeInstanceNameToPID();
                }
                this.mQClientFactory = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQPushConsumer, this.rpcHook);
                this.rebalanceImpl.setConsumerGroup(this.defaultMQPushConsumer.getConsumerGroup());
                this.rebalanceImpl.setMessageModel(this.defaultMQPushConsumer.getMessageModel());
                this.rebalanceImpl.setAllocateMessageQueueStrategy(this.defaultMQPushConsumer.getAllocateMessageQueueStrategy());
                this.rebalanceImpl.setmQClientFactory(this.mQClientFactory);
                this.pullAPIWrapper = new PullAPIWrapper(
                    mQClientFactory,
                    this.defaultMQPushConsumer.getConsumerGroup(), isUnitMode());
                this.pullAPIWrapper.registerFilterMessageHook(filterMessageHookList);
                if (this.defaultMQPushConsumer.getOffsetStore() != null) {
                    this.offsetStore = this.defaultMQPushConsumer.getOffsetStore();
                } else {
                    switch (this.defaultMQPushConsumer.getMessageModel()) {
                        case BROADCASTING:
                            this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
                            break;
                        case CLUSTERING:
                            this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
                            break;
                        default:
                            break;
                    }
                  this.defaultMQPushConsumer.setOffsetStore(this.offsetStore);
                }
                this.offsetStore.load();
                if (this.getMessageListenerInner() instanceof MessageListenerOrderly) {
                    this.consumeOrderly = true;
                    this.consumeMessageService =
                        new ConsumeMessageOrderlyService(this, (MessageListenerOrderly) this.getMessageListenerInner());
                } else if (this.getMessageListenerInner() instanceof MessageListenerConcurrently) {
                    this.consumeOrderly = false;
                    this.consumeMessageService =
                        new ConsumeMessageConcurrentlyService(this, (MessageListenerConcurrently) this.getMessageListenerInner());
                }
                this.consumeMessageService.start();
                boolean registerOK = mQClientFactory.registerConsumer(this.defaultMQPushConsumer.getConsumerGroup(), this);
                if (!registerOK) {
                    this.serviceState = ServiceState.CREATE_JUST;
                    this.consumeMessageService.shutdown();
                    throw new MQClientException("The consumer group[" + this.defaultMQPushConsumer.getConsumerGroup()
                        + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL),
                        null);
                }
                mQClientFactory.start();
                log.info("the consumer [{}] start OK.", this.defaultMQPushConsumer.getConsumerGroup());
                this.serviceState = ServiceState.RUNNING;
                break;
            case RUNNING:
            case START_FAILED:
            case SHUTDOWN_ALREADY:
                throw new MQClientException("The PushConsumer service state not OK, maybe started once, "
                    + this.serviceState
                    + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK),
                    null);
            default:
                break;
        }
        this.updateTopicSubscribeInfoWhenSubscriptionChanged();
        this.mQClientFactory.checkClientInBroker();
        this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
        this.mQClientFactory.rebalanceImmediately();
    }

通过mQClientFactory.start();发我们发现他调用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
java复制代码public void start() throws MQClientException {
        synchronized (this) {
            switch (this.serviceState) {
                case CREATE_JUST:
                    this.serviceState = ServiceState.START_FAILED;
                    // If not specified,looking address from name server
                    if (null == this.clientConfig.getNamesrvAddr()) {
                        this.mQClientAPIImpl.fetchNameServerAddr();
                    }
                    // Start request-response channel
                    this.mQClientAPIImpl.start();
                    // Start various schedule tasks
                    this.startScheduledTask();
                    // Start pull service
                    this.pullMessageService.start();
                    // Start rebalance service
                    this.rebalanceService.start();
                    // Start push service
                  this.defaultMQProducer.getDefaultMQProducerImpl().start(false);
                    log.info("the client factory [{}] start OK", this.clientId);
                    this.serviceState = ServiceState.RUNNING;
                    break;
                case RUNNING:
                    break;
                case SHUTDOWN_ALREADY:
                    break;
                case START_FAILED:
                    throw new MQClientException("The Factory object[" + this.getClientId() + "] has been created before, and failed.", null);
                default:
                    break;
            }
        }
    }

在这个方法中有多个start，我们主要看pullMessageService.start();通过这里我们发现RocketMQ的Push模式底层其实也是通过pull实现的，下面我们来看下pullMessageService处理了哪些逻辑：

1
2
3
4
5
6
7
8
9
java复制代码private void pullMessage(final PullRequest pullRequest) {
        final MQConsumerInner consumer = this.mQClientFactory.selectConsumer(pullRequest.getConsumerGroup());
        if (consumer != null) {
            DefaultMQPushConsumerImpl impl = (DefaultMQPushConsumerImpl) consumer;
            impl.pullMessage(pullRequest);
        } else {
            log.warn("No matched consumer for the PullRequest {}, drop it", pullRequest);
        }
    }

我们发现其实他还是通过DefaultMQPushConsumerImpl类的pullMessage方法来进行消息的逻辑处理.

pullRequest拉取方式

PullRequest这里说明一下，上面我们已经提了一下rocketmq的push模式其实是通过pull模式封装实现的，pullrequest这里是通过长轮询的方式达到push效果。

长轮询方式既有pull的优点又有push模式的实时性有点。

  • push方式是server端接收到消息后，主动把消息推送给client端，实时性高。弊端是server端工作量大，影响性能，其次是client端处理能力不同且client端的状态不受server端的控制，如果client端不能及时处理消息容易导致消息堆积已经影响正常业务等。
  • pull方式是client循环从server端拉取消息，主动权在client端，自己处理完一个消息再去拉取下一个，缺点是循环的时间不好设定，时间太短容易忙等，浪费CPU资源，时间间隔太长client的处理能力会下降，有时候有些消息会处理不及时。
长轮询的方式可以结合两者优点
  1. 检查PullRequest对象中的ProcessQueue对象的dropped是否为true（在RebalanceService线程中为topic下的MessageQueue创建拉取消息请求时要维护对应的ProcessQueue对象，若Consumer不再订阅该topic则会将该对象的dropped置为true）；若是则认为该请求是已经取消的，则直接跳出该方法；
  2. 更新PullRequest对象中的ProcessQueue对象的时间戳（ProcessQueue.lastPullTimestamp）为当前时间戳；
  3. 检查该Consumer是否运行中，即DefaultMQPushConsumerImpl.serviceState是否为RUNNING;若不是运行状态或者是暂停状态（DefaultMQPushConsumerImpl.pause=true），则调用PullMessageService.executePullRequestLater(PullRequest pullRequest, long timeDelay)方法延迟再拉取消息，其中timeDelay=3000；该方法的目的是在3秒之后再次将该PullRequest对象放入PullMessageService. pullRequestQueue队列中；并跳出该方法；
  4. 进行流控。若ProcessQueue对象的msgCount大于了消费端的流控阈值（DefaultMQPushConsumer.pullThresholdForQueue，默认值为1000），则调用PullMessageService.executePullRequestLater方法，在50毫秒之后重新该PullRequest请求放入PullMessageService.pullRequestQueue队列中；并跳出该方法；
  5. 若不是顺序消费（即DefaultMQPushConsumerImpl.consumeOrderly等于false），则检查ProcessQueue对象的msgTreeMap:TreeMap<Long,MessageExt>变量的第一个key值与最后一个key值之间的差额，该key值表示查询的队列偏移量queueoffset；若差额大于阈值（由DefaultMQPushConsumer. consumeConcurrentlyMaxSpan指定，默认是2000），则调用PullMessageService.executePullRequestLater方法，在50毫秒之后重新将该PullRequest请求放入PullMessageService.pullRequestQueue队列中；并跳出该方法；
  6. 以PullRequest.messageQueue对象的topic值为参数从RebalanceImpl.subscriptionInner: ConcurrentHashMap, SubscriptionData>中获取对应的SubscriptionData对象，若该对象为null，考虑到并发的关系，调用executePullRequestLater方法，稍后重试；并跳出该方法；
  7. 若消息模型为集群模式（RebalanceImpl.messageModel等于CLUSTERING），则以PullRequest对象的MessageQueue变量值、type =READ_FROM_MEMORY（从内存中获取消费进度offset值）为参数调用DefaultMQPushConsumerImpl. offsetStore对象（初始化为RemoteBrokerOffsetStore对象）的readOffset(MessageQueue mq, ReadOffsetType type)方法从本地内存中获取消费进度offset值。若该offset值大于0 则置临时变量commitOffsetEnable等于true否则为false；该offset值作为pullKernelImpl方法中的commitOffset参数，在Broker端拉取消息之后根据commitOffsetEnable参数值决定是否用该offset更新消息进度。该readOffset方法的逻辑是：以入参MessageQueue对象从RemoteBrokerOffsetStore.offsetTable:ConcurrentHashMap <MessageQueue,AtomicLong>变量中获取消费进度偏移量；若该偏移量不为null则返回该值，否则返回-1；
  8. 当每次拉取消息之后需要更新订阅关系（由DefaultMQPushConsumer. postSubscriptionWhenPull参数表示，默认为false）并且以topic值参数从RebalanceImpl.subscriptionInner获取的SubscriptionData对象的classFilterMode等于false（默认为false），则将sysFlag标记的第3个字节置为1，否则该字节置为0；
  9. 该sysFlag标记的第1个字节置为commitOffsetEnable的值；第2个字节（suspend标记）置为1；第4个字节置为classFilterMode的值；
  10. 初始化匿名内部类PullCallback，实现了onSucess/onException方法； 该方法只有在异步请求的情况下才会回调；
  11. 调用底层的拉取消息API接口：

PullAPIWrapper.pullKernelImpl

PullAPIWrapper.pullKernelImpl(MessageQueue mq, String subExpression, long subVersion,long offset, int maxNums, int sysFlag,long commitOffset,long brokerSuspendMaxTimeMillis, long timeoutMillis, CommunicationMode communicationMode, PullCallback pullCallback)方法进行消息拉取操作。

将回调类PullCallback传入该方法中，当采用异步方式拉取消息时，在收到响应之后会回调该回调类的方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
java复制代码public void pullMessage(final PullRequest pullRequest) {
        final ProcessQueue processQueue = pullRequest.getProcessQueue();
        if (processQueue.isDropped()) {
            log.info("the pull request[{}] is dropped.", pullRequest.toString());
            return;
        }
        pullRequest.getProcessQueue().setLastPullTimestamp(System.currentTimeMillis());
        try {
            this.makeSureStateOK();
        } catch (MQClientException e) {
            log.warn("pullMessage exception, consumer state not ok", e);
            this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION);
            return;
        }
        if (this.isPause()) {
            log.warn("consumer was paused, execute pull request later. instanceName={}, group={}", this.defaultMQPushConsumer.getInstanceName(), this.defaultMQPushConsumer.getConsumerGroup());
            this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_SUSPEND);
            return;
        }
        long cachedMessageCount = processQueue.getMsgCount().get();
        long cachedMessageSizeInMiB = processQueue.getMsgSize().get() / (1024 * 1024);
        if (cachedMessageCount > this.defaultMQPushConsumer.getPullThresholdForQueue()) {
            this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);
            if ((queueFlowControlTimes++ % 1000) == 0) {
                log.warn(
                    "the cached message count exceeds the threshold {}, so do flow control, minOffset={}, maxOffset={}, count={}, size={} MiB, pullRequest={}, flowControlTimes={}",
                    this.defaultMQPushConsumer.getPullThresholdForQueue(), processQueue.getMsgTreeMap().firstKey(), processQueue.getMsgTreeMap().lastKey(), cachedMessageCount, cachedMessageSizeInMiB, pullRequest, queueFlowControlTimes);
            }
            return;
        }
        if (cachedMessageSizeInMiB > this.defaultMQPushConsumer.getPullThresholdSizeForQueue()) {
            this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);
            if ((queueFlowControlTimes++ % 1000) == 0) {
                log.warn(
                    "the cached message size exceeds the threshold {} MiB, so do flow control, minOffset={}, maxOffset={}, count={}, size={} MiB, pullRequest={}, flowControlTimes={}",
                    this.defaultMQPushConsumer.getPullThresholdSizeForQueue(), processQueue.getMsgTreeMap().firstKey(), processQueue.getMsgTreeMap().lastKey(), cachedMessageCount, cachedMessageSizeInMiB, pullRequest, queueFlowControlTimes);
            }
            return;
        }
        if (!this.consumeOrderly) {
            if (processQueue.getMaxSpan() > this.defaultMQPushConsumer.getConsumeConcurrentlyMaxSpan()) {
                this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_FLOW_CONTROL);
                if ((queueMaxSpanFlowControlTimes++ % 1000) == 0) {
                    log.warn(
                        "the queue's messages, span too long, so do flow control, minOffset={}, maxOffset={}, maxSpan={}, pullRequest={}, flowControlTimes={}",
                        processQueue.getMsgTreeMap().firstKey(), processQueue.getMsgTreeMap().lastKey(), processQueue.getMaxSpan(),
                        pullRequest, queueMaxSpanFlowControlTimes);
                }
                return;
            }
        } else {
            if (processQueue.isLocked()) {
                if (!pullRequest.isLockedFirst()) {
                    final long offset = this.rebalanceImpl.computePullFromWhere(pullRequest.getMessageQueue());
                    boolean brokerBusy = offset < pullRequest.getNextOffset();
                    log.info("the first time to pull message, so fix offset from broker. pullRequest: {} NewOffset: {} brokerBusy: {}",
                        pullRequest, offset, brokerBusy);
                    if (brokerBusy) {
                        log.info("[NOTIFYME]the first time to pull message, but pull request offset larger than broker consume offset. pullRequest: {} NewOffset: {}",
                            pullRequest, offset);
                    }
                    pullRequest.setLockedFirst(true);
                    pullRequest.setNextOffset(offset);
                }
            } else {
                this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION);
                log.info("pull message later because not locked in broker, {}", pullRequest);
                return;
            }
        }
        final SubscriptionData subscriptionData = this.rebalanceImpl.getSubscriptionInner().get(pullRequest.getMessageQueue().getTopic());
        if (null == subscriptionData) {
            this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION);
            log.warn("find the consumer's subscription failed, {}", pullRequest);
            return;
        }
        final long beginTimestamp = System.currentTimeMillis();
        PullCallback pullCallback = new PullCallback() {
            @Override
            public void onSuccess(PullResult pullResult) {
                if (pullResult != null) {
                    pullResult = DefaultMQPushConsumerImpl.this.pullAPIWrapper.processPullResult(pullRequest.getMessageQueue(), pullResult,
                        subscriptionData);
                    switch (pullResult.getPullStatus()) {
                        case FOUND:
                            long prevRequestOffset = pullRequest.getNextOffset();
                            pullRequest.setNextOffset(pullResult.getNextBeginOffset());
                            long pullRT = System.currentTimeMillis() - beginTimestamp;
                            DefaultMQPushConsumerImpl.this.getConsumerStatsManager().incPullRT(pullRequest.getConsumerGroup(),
                                pullRequest.getMessageQueue().getTopic(), pullRT);
                            long firstMsgOffset = Long.MAX_VALUE;
                            if (pullResult.getMsgFoundList() == null || pullResult.getMsgFoundList().isEmpty()) {
                                DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
                            } else {
                                firstMsgOffset = pullResult.getMsgFoundList().get(0).getQueueOffset();
                                DefaultMQPushConsumerImpl.this.getConsumerStatsManager().incPullTPS(pullRequest.getConsumerGroup(),
                                    pullRequest.getMessageQueue().getTopic(), pullResult.getMsgFoundList().size());
                                boolean dispatchToConsume = processQueue.putMessage(pullResult.getMsgFoundList());
                                DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(
                                    pullResult.getMsgFoundList(),
                                    processQueue,
                                    pullRequest.getMessageQueue(),
                                    dispatchToConsume);
                                if (DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval() > 0) {
                                    DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest,
                                        DefaultMQPushConsumerImpl.this.defaultMQPushConsumer.getPullInterval());
                                } else {
                                    DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
                                }
                            }
                            if (pullResult.getNextBeginOffset() < prevRequestOffset
                                || firstMsgOffset < prevRequestOffset) {
                                log.warn(
                                    "[BUG] pull message result maybe data wrong, nextBeginOffset: {} firstMsgOffset: {} prevRequestOffset: {}",
                                    pullResult.getNextBeginOffset(),
                                    firstMsgOffset,
                                    prevRequestOffset);
                            }
                            break;
                        case NO_NEW_MSG:
                            pullRequest.setNextOffset(pullResult.getNextBeginOffset());
                            DefaultMQPushConsumerImpl.this.correctTagsOffset(pullRequest);
                            DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
                            break;
                        case NO_MATCHED_MSG:
                            pullRequest.setNextOffset(pullResult.getNextBeginOffset());
                            DefaultMQPushConsumerImpl.this.correctTagsOffset(pullRequest);
                            DefaultMQPushConsumerImpl.this.executePullRequestImmediately(pullRequest);
                            break;
                        case OFFSET_ILLEGAL:
                            log.warn("the pull request offset illegal, {} {}",
                                pullRequest.toString(), pullResult.toString());
                            pullRequest.setNextOffset(pullResult.getNextBeginOffset());
                            pullRequest.getProcessQueue().setDropped(true);
                            DefaultMQPushConsumerImpl.this.executeTaskLater(new Runnable() {
                                @Override
                                public void run() {
                                    try {
                                        DefaultMQPushConsumerImpl.this.offsetStore.updateOffset(pullRequest.getMessageQueue(),
                                            pullRequest.getNextOffset(), false);
                                        DefaultMQPushConsumerImpl.this.offsetStore.persist(pullRequest.getMessageQueue());
                                        DefaultMQPushConsumerImpl.this.rebalanceImpl.removeProcessQueue(pullRequest.getMessageQueue());
                                        log.warn("fix the pull request offset, {}", pullRequest);
                                    } catch (Throwable e) {
                                        log.error("executeTaskLater Exception", e);
                                    }
                                }
                            }, 10000);
                            break;
                        default:
                            break;
                    }
                }
            }
            @Override
            public void onException(Throwable e) {
                if (!pullRequest.getMessageQueue().getTopic().startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
                    log.warn("execute the pull request exception", e);
                }
                DefaultMQPushConsumerImpl.this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION);
            }
        };
        boolean commitOffsetEnable = false;
        long commitOffsetValue = 0L;
        if (MessageModel.CLUSTERING == this.defaultMQPushConsumer.getMessageModel()) {
            commitOffsetValue = this.offsetStore.readOffset(pullRequest.getMessageQueue(), ReadOffsetType.READ_FROM_MEMORY);
            if (commitOffsetValue > 0) {
                commitOffsetEnable = true;
            }
        }
        String subExpression = null;
        boolean classFilter = false;
        SubscriptionData sd = this.rebalanceImpl.getSubscriptionInner().get(pullRequest.getMessageQueue().getTopic());
        if (sd != null) {
            if (this.defaultMQPushConsumer.isPostSubscriptionWhenPull() && !sd.isClassFilterMode()) {
                subExpression = sd.getSubString();
            }
            classFilter = sd.isClassFilterMode();
        }
        int sysFlag = PullSysFlag.buildSysFlag(
            commitOffsetEnable, // commitOffset
            true, // suspend
            subExpression != null, // subscription
            classFilter // class filter
        );
        try {
            // 下面我们看继续跟进这个方法，这个方法已经就是客户端如何拉取消息
            this.pullAPIWrapper.pullKernelImpl(
                pullRequest.getMessageQueue(),
                subExpression,
                subscriptionData.getExpressionType(),
                subscriptionData.getSubVersion(),
                pullRequest.getNextOffset(),
                this.defaultMQPushConsumer.getPullBatchSize(),
                sysFlag,
                commitOffsetValue,
                BROKER_SUSPEND_MAX_TIME_MILLIS,
                CONSUMER_TIMEOUT_MILLIS_WHEN_SUSPEND,
                // 消息的通信方式为异步
                CommunicationMode.ASYNC,
                pullCallback
            );
        } catch (Exception e) {
            log.error("pullKernelImpl exception", e);
            this.executePullRequestLater(pullRequest, PULL_TIME_DELAY_MILLS_WHEN_EXCEPTION);
        }
    }

发送远程请求拉取消息

在MQClientAPIImpl.pullMessage方法中，根据入参communicationMode的值分为异步拉取和同步拉取方式两种。

无论是异步方式拉取还是同步方式拉取，在发送拉取请求之前都会构造一个ResponseFuture对象，以请求消息的序列号为key值，存入NettyRemotingAbstract.responseTable:ConcurrentHashMap, ResponseFuture>变量中，对该变量有几种情况会处理：

  1. 发送失败后直接删掉responseTable变量中的相应记录；
  2. 收到响应消息之后，会以响应消息中的序列号（由服务端根据请求消息的序列号原样返回）从responseTable中查找ResponseFuture对象，并设置该对象的responseCommand变量。若是同步发送会唤醒等待响应的ResponseFuture.waitResponse方法；若是异步发送会调用ResponseFuture.executeInvokeCallback()方法完成回调逻辑处理；
  3. 在NettyRemotingClient.start()启动时，也会初始化定时任务，该定时任务每隔1秒定期扫描responseTable列表，遍历该列表中的ResponseFuture对象，检查等待响应是否超时，若超时，则调用ResponseFuture. executeInvokeCallback()方法，并将该对象从responseTable列表中删除；
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
java复制代码public PullResult pullMessage(
        final String addr,
        final PullMessageRequestHeader requestHeader,
        final long timeoutMillis,
        final CommunicationMode communicationMode,
        final PullCallback pullCallback
    ) throws RemotingException, MQBrokerException, InterruptedException {
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, requestHeader);
        switch (communicationMode) {
            case ONEWAY:
                assert false;
                return null;
            case ASYNC:
                this.pullMessageAsync(addr, request, timeoutMillis, pullCallback);
                return null;
            case SYNC:
                return this.pullMessageSync(addr, request, timeoutMillis);
            default:
                assert false;
                break;
        }
        return null;
    }

同步拉取

对于同步发送方式，调用MQClientAPIImpl.pullMessageSync(String addr, RemotingCommand request, long timeoutMillis)方法,大致步骤如下：

  1. 调用RemotingClient.invokeSync(String addr, RemotingCommand request, long timeoutMillis)方法：
    • 获取Broker地址的Channel信息。根据broker地址从RemotingClient.channelTables:ConcurrentHashMap, ChannelWrapper>变量中获取ChannelWrapper对象并返回该对象的Channel变量；若没有ChannelWrapper对象则与broker地址建立新的连接并将连接信息存入channelTables变量中，便于下次使用；
    • 若NettyRemotingClient.rpcHook:RPCHook变量不为空（该变量在应用层初始化DefaultMQPushConsumer或者DefaultMQPullConsumer对象传入该值），则调用RPCHook.doBeforeRequest(String remoteAddr, RemotingCommand request)方法；
    • 调用NettyRemotingAbstract.invokeSyncImpl(Channel channel, RemotingCommand request, long timeoutMillis)方法，该方法的逻辑如下：
      • A）使用请求的序列号（opaue）、超时时间初始化ResponseFuture对象；并将该ResponseFuture对象存入NettyRemotingAbstract.responseTable: ConcurrentHashMap变量中；
      • B）调用Channel.writeAndFlush(Object msg)方法将请求对象RemotingCommand发送给Broker；然后调用addListener(GenericFutureListener<? extends Future<? super Void>> listener)方法添加内部匿名类：该内部匿名类实现了ChannelFutureListener接口的operationComplete方法，在发送完成之后回调该监听类的operationComplete方法，在该方法中，首先调用ChannelFuture. isSuccess()方法检查是否发送成功，若成功则置ResponseFuture对象的sendRequestOK等于true并退出此回调方法等待响应结果；若不成功则置ResponseFuture对象的sendRequestOK等于false，然后从NettyRemotingAbstract.responseTable中删除此请求序列号（opaue）的记录，置ResponseFuture对象的responseCommand等于null，并唤醒ResponseFuture.waitResponse(long timeoutMillis)方法的等待；
      • C）调用ResponseFuture.waitResponse(long timeoutMillis)方法等待响应结果；在发送失败或者收到响应消息（详见5.10.3小节）或者超时的情况下会唤醒该方法返回ResponseFuture.responseCommand变量值；
      • D）若上一步返回的responseCommand值为null，则抛出异常:若ResponseFuture.sendRequestOK为true，则抛出RemotingTimeoutException异常，否则抛出RemotingSendRequestException异常；
      • E）若上一步返回的responseCommand值不为null，则返回responseCommand变量值；
    • 若NettyRemotingClient.rpcHook: RPCHook变量不为空，则调用RPCHook.doAfterResponse(String remoteAddr, RemotingCommand request)方法；
  • 以上一步的返回值RemotingCommand对象为参数调用MQClientAPIImpl. processPullResponse (RemotingCommand response)方法将返回对象解析并封装成PullResultExt对象然后返回给调用者，响应消息的结果状态转换如下：
    • 若RemotingCommand对象的Code等于SUCCESS，则PullResultExt.pullStatus=FOUND；
    • 若RemotingCommand对象的Code等于PULL_NOT_FOUND，则PullResultExt.pullStatus= NO_NEW_MSG；
    • 若RemotingCommand对象的Code等于PULL_RETRY_IMMEDIATELY，则PullResultExt.pullStatus= NO_MATCHED_MSG；
    • 若RemotingCommand对象的Code等于PULL_OFFSET_MOVED，则PullResultExt.pullStatus= OFFSET_ILLEGAL；
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
java复制代码@Override
    public RemotingCommand invokeSync(String addr, final RemotingCommand request, long timeoutMillis)
        throws InterruptedException, RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException {
        long beginStartTime = System.currentTimeMillis();
        final Channel channel = this.getAndCreateChannel(addr);
        if (channel != null && channel.isActive()) {
            try {
                if (this.rpcHook != null) {
                    this.rpcHook.doBeforeRequest(addr, request);
                }
                long costTime = System.currentTimeMillis() - beginStartTime;
                if (timeoutMillis < costTime) {
                    throw new RemotingTimeoutException("invokeSync call timeout");
                }
                RemotingCommand response = this.invokeSyncImpl(channel, request, timeoutMillis - costTime);
                if (this.rpcHook != null) {
                    this.rpcHook.doAfterResponse(RemotingHelper.parseChannelRemoteAddr(channel), request, response);
                }
                return response;
            } catch (RemotingSendRequestException e) {
                log.warn("invokeSync: send request exception, so close the channel[{}]", addr);
                this.closeChannel(addr, channel);
                throw e;
            } catch (RemotingTimeoutException e) {
                if (nettyClientConfig.isClientCloseSocketIfTimeout()) {
                    this.closeChannel(addr, channel);
                    log.warn("invokeSync: close socket because of timeout, {}ms, {}", timeoutMillis, addr);
                }
                log.warn("invokeSync: wait response timeout exception, the channel[{}]", addr);
                throw e;
            }
        } else {
            this.closeChannel(addr, channel);
            throw new RemotingConnectException(addr);
        }
    }

getMQClientAPIImpl().pullMessage最终通过channel写入并刷新队列中。然后在消息服务端大体的处理逻辑是服务端收到新消息请求后，如果队列中没有消息不急于返回，通过一个循环状态，每次waitForRunning一段时间默认5秒，然后再check，如果broker一直没有新新消息，第三次check的时间等到时间超过SuspendMaxTimeMills就返回空，如果在等待过程中收到了新消息直接调用notifyMessageArriving函数返回请求结果。“长轮询”的核心是，Broker端HOLD住客户端过来的请求一小段时间，在这个时间内有新消息到达，就利用现有的连接立刻返回消息给 Consumer 。长轮询的主动权掌握在consumer中，即使broker有大量的消息堆积也不会主动推送给consumer。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

「这是我参与11月更文挑战的第 15 天，活动详情查看：2021最后一次更文挑战」

错误处理

你可能已经注意到，在整个代码中，一直都在引用一个 SQLRiteError 类型。这是我使用 thiserror crate 定义的错误类型，它很容易使用的库，为标准库的 std::error::Error 特质提供了一个方便的衍生宏。不过提到我遇到了这个trait，它基本上解决了很多问题，代码看起来超级干净！目前的错误模块，位于 src/error.rs 中。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
rust复制代码use thiserror::Error;

use sqlparser::parser::ParserError;

// 别名，便于书写，偷个懒～～～
pub type Result<T> = result::Result<T, SQLRiteError>;

#[derive(Error, Debug, PartialEq)]
pub enum SQLRiteError {
    #[error("Not Implemented error: {0}")]
    NotImplemented(String),
    #[error("General error: {0}")]
    General(String),
    #[error("Internal error: {0}")]
    Internal(String),
    #[error("Unknown command error: {0}")]
    UnknownCommand(String),
    #[error("SQL error: {0:?}")]
    SqlError(#[from] ParserError),
}

/// Return SQLRite errors from String
pub fn sqlrite_error(message: &str) -> SQLRiteError {
    SQLRiteError::General(message.to_owned())
}

简单介绍一下

Rust界的大佬 dtolnay 设计了两个crate：thiserror 和 anyhow。在这之前已经有很多提升错误处理的库了，但最后最终还是解决 thiserror + anyhow 是最易用和实用的。

注意事项：thiserror 是给lib使用的，而 anyhow 是给bin程序使用，当然bin程序可以使用thiser+anyhow，但是切记 anyhow 不要在lib里面使用 。

而我们在上面的 SQLRiteError 定义中，使用的就是 thiserror 的包装，其实就是使用 enum 封装的思路。

1
2
3
4
5
6
rust复制代码use anyhow::Result; 
fn get_cluster_info() -> Result<ClusterMap> { 
    let config = std::fs::read_to_string("cluster.json")?; 
    let map: ClusterMap = serde_json::from_str(&config)?; 
    Ok(map) 
}

而 anyhow 使用上，直接返回 anyhow::Result，因为它实现了 From<E> where E: Error，所以就不用再维护一个 enum 类型。

暂时总结

目前我们成功地解析了用户的命令，区分了 MetaCommand 和 SQLCommand。以及实现了一个可扩展的MetaCommand 模块，使得将来增加更多的命令变得容易。

然后添加了sql模块，通过使用 sqlparser-rs crate，我们成功地解析了SQL语句，并能够从每个SQL语句中生成一个 ast。我们已经从 CREATE TABLE SQL 语句中解析并生成了至少一个简化版的字节码，可以进入数据库(将在后面进行)。最后，我们还创建了一个错误模块，所以我们有一个标准化的方法来处理整个应用程序的错误。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

「这是我参与11月更文挑战的第 21 天，活动详情查看：2021最后一次更文挑战」

点我进入 minikube 试炼

今天我们先来尝试使用一下 minikube ，可以进入到 kubernetes.io/zh/docs/tut… 页面上直接感受，或者通过如下指令，将 minikube 放入我们的服务器上面进行使用

简单安装 minikube

Linux 的

curl -LO https://storage.googleapis.com/minikube/releases/latest/minikube-linux-amd64 sudo install minikube-linux-amd64 /usr/local/bin/minikube

windows 的

New-Item -Path 'c:\' -Name 'minikube' -ItemType Directory -Force Invoke-WebRequest -OutFile 'c:\minikube\minikube.exe' -Uri 'https://github.com/kubernetes/minikube/releases/latest/download/minikube-windows-amd64.exe' -UseBasicParsing

例如

运行我们的集群

minikube start

启动集群

minikube pause

在不影响部署的应用程序的情况下暂停 Kubernete

1
2
3
shell复制代码$ minikube pause
* Pausing node m01 ... 
* Paused 20 containers in: kube-system, kubernetes-dashboard, storage-gluster, istio-operator

此时 minikube 已经暂停了，我们查看 pod 列表是查看不了的

minikube unpause

取消暂停的实例

1
2
3
shell复制代码$ minikube unpause
* Unpausing node m01 ... 
* Unpaused 20 containers in: kube-system, kubernetes-dashboard, storage-gluster, istio-operator

minikube stop

停止 minikue

minikube addons list

列出当前支持的插件

kubectl config view

查看 kubectl 的配置

kubectl get pod -A

列出所有命名空间的对象

部署一个应用

1
2
shell复制代码$ kubectl  create deployment hello-xiaomotong --image=k8s.gcr.io/echoserver:1.4
deployment.apps/hello-xiaomotong created

我们可以看到，hello-xiaomotong 已经部署到集群中了

  • NAMESPACE

命名空间

  • NAME

应用名称

  • READY

表示该 pod 可以为请求提供服务，并且应该被添加到对应服务的负载均衡池中，关于此处还有其他的标识，后续详细写到 pod 的时候，我们可以详细学习

  • STATUS

状态，此时是 Running 状态，正常运行

对外暴露服务端口

  • 对外暴露服务端口，暴露 9999 端口
1
2
shell复制代码$ kubectl expose deployment hello-xiaomotong --type=NodePort --port=9999
service/hello-xiaomotong exposed

kubectl get service

查看 service 信息

可以看到，我们的外部 9999 端口，映射到服务内部 32403 端口上，那么现在我们外部访问服务的 9999 端口，就可以访问到这个 pod 内部的服务了

k8S 的 3 钟外部访问方式

此处我们可以看到 TYPE 字段 ， 目前可以有 2 种类型，实际上这个是 k8S 的外部访问方式，一共有 3 种：

  • LoadBalancer
  • NodePort
  • Ingress

上述 3 种方式，都是将集群外部流量导入到集群内的方式，只是实现方式不同

ClusterIP 是 K8S 集群内部的默认服务，集群内的其它应用都可以访问该服务，但是集群外部无法访问它 ，

如果需要外部访问 ClusterIP 类型的服务，也是可以的，需要加一个代理 ， 我们后续可以详细说一下上述 几种方式的使用方式，场景，优缺点等等

删除集群

1
2
3
shell复制代码minikube delete

minikube delete  --all

感兴趣的小伙伴可以 点我查看 monikube 手册

今天就到这里，学习所得，若有偏差，还请斧正

欢迎点赞，关注，收藏

朋友们，你的支持和鼓励，是我坚持分享，提高质量的动力

好了，本次就到这里

技术是开放的，我们的心态，更应是开放的。拥抱变化，向阳而生，努力向前行。

我是小魔童哪吒，欢迎点赞关注收藏，下次见~

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

「这是我参与11月更文挑战的第2天，活动详情查看：2021最后一次更文挑战」

一、连接池知识背景

在我们的程序中连接数据库是一种耗时的行为，.NET为了降低打开连接的成本，在ado.net中使用了一种叫做连接池的优化技术。使用数据库连接池可以减少打开新连接的次数，并且将物理数据库的连接交给了池程序去做。
池程序是通过为每个特定的连接配置保持一组活动的连接对象来管理数据库连接的。每当应用程序发起连接数据库的请求时，池程序就会在连接池中查找是否存在可用的连接，如果有则返回给调用者。当应用程序关闭连接对象时，池程序将连接对象返回到池中, 这个连接可以在下一次发起连接数据库时重用。
那么.NET是如何形成数据库连接池的呢？首先只有相同的连接配置才能被池化，.NET为不同的配置维护了不同的连接池。这里所说的相同配置必须具有相同的进程、相同的连接字符串以及连接字符串关键key顺序相同。连接池中可用连接数量是由连接字符串中的Max Pool Size决定的。例如在一个应用程序中数据库连接相关的代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
csharp复制代码using (SqlConnection connection = new SqlConnection("Integrated Security=SSPI;Initial Catalog=test1"))  
{  
	connection.Open();
} 
using (SqlConnection connection = new SqlConnection("Integrated Security=SSPI;Initial Catalog=test2"))  
{  
	connection.Open();
}  
  
using (SqlConnection connection = new SqlConnection("Integrated Security=SSPI;Initial Catalog=test1"))  
{  
	connection.Open();
}

在上面的代码中虽然创建了三个Connection对象，但是却只形成了两个数据库连接池。那么连接池中的连接什么时候会被移除呢？答案是连接池中的连接空闲4-8 分钟后就会被池程序会移除，或者是应用程序进程关闭连接池中的连接也会被移除。

二、清空.NET连接池

前面简单守卫说了一下连接池相关的内容，现在我们就来看一下如何清空数据库连接池。
在.NET中提供了ClearAllPools和ClearPool静态方法用于清空连接池。其中ClearAllPools表示清空与指定的DBProvider相关的所有数据库连接池，ClearPool(DBConnection conn)表示清空与指定连接对象相关的连接池。一般来说我们常用的是ClearPool(DBConnection conn) 方法。下面我们就使用ClearPool方法来演示一下如何清空数据库连接池：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
csharp复制代码public class DBHelper
{
    public string Get()
    {
        var s = "User ID=root;Password=1qazxsw2;DataBase=test;Server=127.0.0.1;Port=6987;Min Pool Size=5;Max Pool Size=50;CharSet=utf8;";
        using (var conn = new MySqlConnection(s))
        {
            var comm = conn.CreateCommand();
            comm.CommandText = "select count(*) from usertest;";
            conn.Open();
            var ret = comm.ExecuteScalar();
            comm.CommandText = "select count(*) from TestTable;";
            var len = comm.ExecuteScalar();
            return $"查询结果为:{ret} ,连接池的连接对象数量为: {len}";
        };
    }
    public string CP()
    {
        var s = "User ID=root;Password=1qazxsw2;DataBase=test;Server=127.0.0.1;Port=6987;Min Pool Size=5;Max Pool Size=50;CharSet=utf8;";
        using (var conn = new MySqlConnection(s))
        {
            conn.Open();
            MySqlConnection.ClearPool(conn);
        };
        using (var conn = new MySqlConnection(s))
        {
            conn.Open();
            var comm = conn.CreateCommand();
            comm.CommandText = "select count(*) from TestTable;";
            var len = comm.ExecuteScalar();
            return $"连接池已经清空， 当前查询连接池对象有 {ken}  个";
        }
    }
}

我们在main函数中调用上面的代码，并查询mysql会发现数据库连接池的数据和我们上面代码执行的结果是一样的。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

这是我参与11月更文挑战的第21天，活动详情查看：2021最后一次更文挑战。

题目描述：

你总共有 n 枚硬币，并计划将它们按阶梯状排列。对于一个由 k 行组成的阶梯，其第 i 行必须正好有 i 枚硬币。阶梯的最后一行 可能 是不完整的。

给你一个数字 n ，计算并返回可形成 完整阶梯行 的总行数。

示例一

image.png

1
2
3
ini复制代码输入： n = 5
输出： 2
解释： 因为第三行不完整，所以返回 2 。

示例二

image.png

1
2
3
ini复制代码输入： n = 8
输出： 3
解释： 因为第四行不完整，所以返回 3 。

提示：

  • 1 <= n <= 2^31 - 1

思路分析

二分法

我们知道第 i 行必须 正好 有 i 个硬币（最后一行除外），再根据数学法，我们知道，到第 i 行时总共使用的硬币数量为 total=i(i+1)/2

所以现在我们的目标就是要寻找这么一个 i 使用得 total 小于或等于 n，而且 i 介于 1 和 n 之间。

除去暴力法，我们很容易就想到二分法了。

这里我们假设左边界为 1，右边界为 n,这里取中的时候有个小细节，我们要让结果更靠近右边界，所有有个 +1 的操作，直接上代码。

AC代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
Kotlin复制代码class Solution {
    fun arrangeCoins(n: Int): Int {
        var left = 1
        var right = n
        while(left < right) {
            val mid = (right - left + 1)/2 + left
            if(mid.toLong() * (mid + 1) / 2 <= n ) {
                left = mid
            }else {
                right = mid - 1
            }
        }
        return left
    }
}

总结

虽然是二分法，但是这里的 +1 非常的关键，平常的二分法其实偏左偏右都可以，但是这题要取偏右的，不然如果就剩2个点的时候，mid = left，就会死循环，如果不 +1 ，这一题的答案也会因为死循环超时的。

参考

排列硬币 - 排列硬币 - 力扣（LeetCode） (leetcode-cn.com)

【宫水三叶】一题双解 :「数学」&「二分」 - 排列硬币 - 力扣（LeetCode） (leetcode-cn.com)

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

这是我参与11月更文挑战的第21天，活动详情查看：2021最后一次更文挑战

前言

在我们使用C++来写算法时经常会使用到algorithm库，下面我们来通过一个题目来简单展示一下这个函数的功能。

题目

读入 n（>0）名学生的姓名、学号、成绩，分别输出成绩最高和成绩最低学生的姓名和学号。

输入格式：

每个测试输入包含 1 个测试用例，格式为

第 1 行：正整数 n
第 2 行：第 1 个学生的姓名 学号 成绩
第 3 行：第 2 个学生的姓名 学号 成绩
… … …
第 n+1 行：第 n 个学生的姓名 学号 成绩

其中姓名和学号均为不超过 10 个字符的字符串，成绩为 0 到 100 之间的一个整数，这里保证在一组测试用例中没有两个学生的成绩是相同的。

输出格式：

对每个测试用例输出 2 行，第 1 行是成绩最高学生的姓名和学号，第 2 行是成绩最低学生的姓名和学号，字符串间有 1 空格。

输入样例：

3
Joe Math990112 89
Mike CS991301 100
Mary EE990830 95

结尾无空行

输出样例：

Mike CS991301
Joe Math990112

结尾无空行

思路

使用algorithm库中的sort函数来进行快速排序在按要求输出内容

sort函数

sort(a,b,c)

a : 数组头指针

b: 数组头指针+数组长度

c:比较函数指针

注意：sort数组采用了前闭后开的集合来存放数组

代码

1、建立一个结构体来记录学生的各种属性

struct Student
{
char name[11];
char number[11];
int great;
} student[N];

2、创建一个比较函数

bool cmp(Student a,Student b){
return a.great > b.great;
}

3、调用sort函数进行比较

sort(student,student+n,cmp);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
c++复制代码#include<iostream>
#include<algorithm>
using namespace std;
const int N = 1e6 + 10;
int n;
struct Student
{
    char name[11];
    char number[11];
    int great;
} student[N];
bool cmp(Student a,Student b){
    return a.great > b.great;
}
int main(){
    scanf("%d",&n);
    for (int i = 0; i < n; i++)
    {
        scanf("%s %s %d",student[i].name,student[i].number,&student[i].great);
    }
    sort(student,student+n,cmp);
    
    cout << student[0].name <<" " <<  student[0].number << endl;
    cout << student[n-1].name <<" " <<  student[n-1].number << endl;
    
    return 0;
}

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

「这是我参与11月更文挑战的第20天，活动详情查看：2021最后一次更文挑战」

相关文章

Java随笔记：Java随笔记

前言

paths()过滤

  • 首先看看需要什么参数
+ ![image-20211122220925409.png](https://gitee.com/songjianzaina/juejin_p7/raw/master/img/4bb79128ea49ef19e63a8e6066909a92df94ca9de80f071ec9d34a18f836b0ce)
+ 点进去看看
+ 
1
2
3
4
kotlin复制代码    public ApiSelectorBuilder paths(Predicate<String> selector) {
        this.pathSelector = Predicates.and(this.pathSelector, selector);
        return this;
    }

①、PathSelectors.any()

  • 一样的，我们先举个例子看看
+ 
1
2
3
4
5
6
7
8
scss复制代码 @Bean
    public Docket docket(){
        return new Docket(DocumentationType.SWAGGER_2)
                .apiInfo(apiInfo())
                .select()
                .paths(PathSelectors.any())
                .build();
    }


+ ![image-20211122221444841.png](https://gitee.com/songjianzaina/juejin_p7/raw/master/img/81b1a7c97a0183a08fb915863c4413bf60b6ad088e3597324130e4f1695dcb29)
+ 重启项目看看
+ ![image-20211122221545354.png](https://gitee.com/songjianzaina/juejin_p7/raw/master/img/b583cd1c252fc3f72cabe941a678b62837fb3530ec12852d716bbbd183ea99c3)
+ 可以看到，包括`error`的controller全部被扫描出来了。
  • 然后我们再进PathSelectors看看，这是什么玩意？
+ 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
typescript复制代码public class PathSelectors {
    private PathSelectors() {
        throw new UnsupportedOperationException();
    }
​
    public static Predicate<String> any() {
        return Predicates.alwaysTrue();
    }
​
    public static Predicate<String> none() {
        return Predicates.alwaysFalse();
    }
​
    public static Predicate<String> regex(final String pathRegex) {
        return new Predicate<String>() {
            public boolean apply(String input) {
                return input.matches(pathRegex);
            }
        };
    }
​
    public static Predicate<String> ant(final String antPattern) {
        return new Predicate<String>() {
            public boolean apply(String input) {
                AntPathMatcher matcher = new AntPathMatcher();
                return matcher.match(antPattern, input);
            }
        };
    }
}


+ 可以看出一共四个参数：any()、none()、regex()、ant()
+ 由此可以先得出第一个结论：any() 不论啥，我全都要！
  • 我们每种都玩一玩，最后再去总结每种的区别。

②、PathSelectors.none()

  • 1
    2
    3
    4
    5
    6
    7
    8
    scss复制代码    @Bean
        public Docket docket(){
            return new Docket(DocumentationType.SWAGGER_2)
                    .apiInfo(apiInfo())
                    .select()
                    .paths(PathSelectors.none())
                    .build();
        }

  • 重启看结果

  • image-20211122221959073.png

  • 很明显了，所有的都没被展示！我全都不要！

③、PathSelectors.regex()

  • 1
    2
    3
    4
    5
    6
    7
    8
    scss复制代码    @Bean
        public Docket docket(){
            return new Docket(DocumentationType.SWAGGER_2)
                    .apiInfo(apiInfo())
                    .select()
                    .paths(PathSelectors.regex("^[+-@=](.*?)"))//该正则表示匹配所有
                    .build();
        }

  • 重启看结果

  • image-20211122223432013.png

  • 1
    2
    3
    4
    5
    6
    7
    8
    scss复制代码    @Bean
        public Docket docket(){
            return new Docket(DocumentationType.SWAGGER_2)
                    .apiInfo(apiInfo())
                    .select()
                    .paths(PathSelectors.regex("(/test)([+-@=])(.*?)"))//Java正则表达式以括号分组，表示匹配以/test开头的所有controller
                    .build();
        }

  • 重启看结果

  • image-20211122223806247.png

  • 这样的话结果就很明显啦！根据正则表达式来过滤哪些需要展示，哪些不需要！

  • 关于正则表达式的话，后面会单独写点文章来玩一玩的！这里按下不表！

④、PathSelectors.ant()

  • 1
    2
    3
    4
    5
    6
    7
    8
    scss复制代码    @Bean
        public Docket docket(){
            return new Docket(DocumentationType.SWAGGER_2)
                    .apiInfo(apiInfo())
                    .select()
                    .paths(PathSelectors.ant("/test-swagger2/**"))//匹配/test-swagger2/开头的所有controller
                    .build();
        }

  • 重启看结果

  • image-20211122224117121.png

  • 这个也不做过多展示，只提一个小技巧。

  • 当我们项目的mapping以功能划分时：

+ ![image-20211122224232017.png](https://gitee.com/songjianzaina/juejin_p7/raw/master/img/4465d72695c9259de8c87ff2039095f1b04e2f171caba4b1b00625244723e02e)
+ 即test功能下所有的都被扫描到！
+ 而用户功能我不加进去，那么即不会被扫描到！

总结

  • any() // 任何请求都扫描
  • none() // 任何请求都不扫描
  • regex(final String pathRegex) // 通过正则表达式控制
  • ant(final String antPattern) // 通过ant()控制
  • 明天带来的就是Swagger的开关怎么玩？如何控制多环境的开关和闭合！
  • 以上内容都是个人见解，如有不对，敬请指出！

路漫漫其修远兮，吾必将上下求索~

如果你认为i博主写的不错！写作不易，请点赞、关注、评论给博主一个鼓励吧~hahah

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

👨‍🎓作者：bug菌

✏️博客：CSDN、掘金、infoQ、51CTO等

🎉简介：CSDN博客专家，C站历届博客之星Top50，掘金/InfoQ/51CTO等社区优质创作者，全网合计8w粉+，对一切技术感兴趣，重心偏Java方向；硬核公众号「 猿圈奇妙屋」，欢迎小伙伴们的加入，一起秃头，一起变强。

..

✍️温馨提醒：本文字数：999字, 阅读完需：约 5 分钟

嗨，家人们，我是bug菌呀，我又来啦。今天我们来聊点什么咧，OK，接着为大家更[《springboot零基础入门教学》](https://blog.csdn.net/weixin_43970743/category_11599389.html)系列文章吧。希望能帮助更多的初学者们快速入门！

小伙伴们在批阅文章的过程中如果觉得文章对您有一丝丝帮助，还请别吝啬您手里的赞呀，大胆的把文章点亮👍吧，您的点赞三连(收藏⭐+关注👨‍🎓+留言📃)就是对bug菌我创作道路上最好的鼓励与支持😘。时光不弃🏃🏻‍♀️，创作不停💕，加油☘️

一、前言

实际的项目开发中，一个项目通常会存在多个环境，例如，开发环境、测试环境和生产环境等。不同环境的配置也不尽相同，例如开发环境使用的是开发数据库，测试环境使用的是测试数据库，而生产环境使用的是线上的正式数据库。

所以问题来了，每次发布测试环境或者上生产，配置环境怎么配置？对吧？不要急，springboot都帮我们做了，我们只需要会使用profile就好！

早在Spring3.1版本时，profile就已经出来了。profile 是什么？它就是可以让 Spring 对不同的环境提供不同配置的功能，可以通过激活、指定参数等方式快速切换环境。

**简言之：**就是我们需要在不同的场景下使用不同的配置，profile就是为解决我们多环境下切换配置复杂的问题而诞生的。

二、如何使用profile？

Spring Boot 的配置文件共有两种形式：.properties 文件和 .yml 文件，不管哪种形式，它们都能通过文件名的命名形式区分出不同的环境的配置，文件命名格式为：

1
bash复制代码application-{profile}.properties/yaml

其中，{profile} 一般为各个环境的名称或简称，例如 dev、test 和 pro 等等。如下我们就直接使用yaml配置文件格式来做演示：

yaml 配置

在 demo项目下 的 config文件夹 下添加 如下4 个配置文件：

  • application.yaml：主配置文件
  • application-dev.yaml：开发环境配置文件
  • application-test.yaml：测试环境配置文件
  • application-pro.yaml：生产环境配置文件

在 application.yaml 文件中，指定默认服务器端口号为 8080，并通过以下配置激活开发环境（dev）的 profile。

未指定环境时，启动项目，你们可以看到控制台打印如下：

1
vbnet复制代码No active profile set, falling back to default profiles: default

看我圈起来的内容，很明显是有口子可以进行配置的。那应该得如何使用呢？别着急，往下看。

我们来做个试验吧，我们在核心配置文件application.yaml中设置属性。spring.profiles.active=dev，再启动程序，发现application-dev.yaml被激活了，启动端口号也修改为子配置文件设置的端口号。

看控制台打印，测试可知：

很明显可以看到dev环境已经启动了。

有小伙伴可能会不相信，该不会是巧合吧，那我再给大家做个试验，这次在test环境中配置端口为8090，再请大家看结果会是怎样？

application-test.yaml 内容设置如下：

1
2
yaml复制代码server:
  port: 8090

application.yaml 内容设置如下：

1
2
3
yaml复制代码spring:
    profiles:
      active: test

最终它到底能否启动项目成功并且运行8090端口呢？还是运行未指定环境的8080端口呢？接着往下看；

ok！大家请看，确实是运行的tests环境配置。如此，便证实在两次application.yaml测试中使用profileh成功动态切换配置。

上述介绍的主要是配置方式动态切换，而切换方式就是通过配置文件的spring.profiles.active属性实现，那还有没有别的启动方式呢？别着急，肯定有，看下边，bug菌都给大家总结好了，可得好好学哈~

如下两种激活配置文件的方式：

1
2
3
4
5
ini复制代码profile激动方式配置文件：
在配置文件中配置：spring.profiles.active = dev(如上已经做了演示)

虚拟机参数配置：在VM options指定：-Dspring.profiles.active = dev
命令行参数配置：java -jar xxx.jar  --spring.profiles.active = dev

然后就是.properties文件配置方式也跟.yaml是一样的，此处就不一一赘述啦。各位小伙伴大可自行尝试，产生任何疑惑的地方都可下方直接留言哦~bug菌一定不留余力的帮助大家。赠人玫瑰，手有余香。

接下来 我就为大家介绍及使用剩下的两种激活配置的方式

三、虚拟机参数方式配置

在idea开发工具中打开RunDebug Configurations界面；然后在VM options一行中指定；

1
ini复制代码-Dspring.profiles.active=dev

说明：指定是dev环境；dev换成别的环境也可。就=后边改成别的环境即可。

application-dev.yaml

1
yaml复制代码server:  port: 8100

项目启动一切正常。说明VM参数配置动态切换环境可行。

四、命令行参数配置

先是将我们的demo项目打成一个jjar包

先教下大家怎么使用idea打jar包吧，其实很简单，看如下步骤瞬间就可以学会。

1
csharp复制代码[INFO] BUILD SUCCESS

打印出了 success ，说明jar打包成功；

接着我们找到jar的位置；看如下图：是存到了target根目录下

然后选中该jar文件，右键单击打开面板选择Show in Explorer 即可跳转到该文件位置上，

如下已经是找到了该文件存放位置，接着直接在目录上输入cmd回车，即可打开小黑框。

在小黑框里执行如下命令，看看能否运行成功呢？咱们拭目以待；

1
ini复制代码java -jar demo-0.0.1-SNAPSHOT.jar --spring.profiles.active=test

说明：–spring.profiles.active=test 为激活测试环境（test）Profile 的命令行参数。

运行结果如下：

ok，完美，成功运行！证明命令行参数配置也可动态切换环境。

至于大家如何选择，这就看你的实际需求啦！

OK，以上就是这期所有的内容啦，如果有任何问题欢迎评论区批评指正，咱们下期见。

五、热文推荐：

文末🔥

如果还想要学习更多，小伙伴们可关注bug菌专门为大家创建的专栏《springboot零基础入门教学》，从无到有，从零到一！希望能帮助到更多小伙伴们。

我是bug菌，一名想走👣出大山改变命运的程序猿。接下来的路还很长，都等待着我们去突破、去挑战。来吧，小伙伴们，我们一起加油！未来皆可期，fighting！

感谢认真读完我博客的铁子萌，在这里呢送给大家一句话，不管你是在职还是在读，绝对终身受用。

时刻警醒自己：

抱怨没有用，一切靠自己；

想要过更好的生活，那就要逼着自己变的更强，生活加油！！！

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

发表于

「这是我参与11月更文挑战的第20天，活动详情查看：2021最后一次更文挑战」

项目开发中，存在很多的业务需求要求对页面的相关数据进行导入和导出，而对于数据量较大的页面又要考虑在导出时不能影响到服务运行的稳定性，因此在选择数据导入导出框架时要考虑业务的需求。

对于低内存占用导出大量数据的情景、EasyExcel框架会有很好的效果实现。

  1. EaseExcel

1.1 EaseExcel介绍

EasyExcel是一个基于Java的简单、省内存读写Excel的开源项目，EasyExcel是在Apache poi框架的基础上对其进行深度优化，减少Excel文件导入导出时对系统内存的占用，实现尽可能少的使用内存来完成上百M大小Excel文件的读写。

Apache poi在使用时会首先将文件加载到内存中进行操作，因此对于几十上百M的文件有可能会发生OOM内存溢出导致服务不可用，而EasyExcel优化可以实现20s读取75M文件只占用64M内存，有效的避免了OOM的发生。

1.2 EaseExcel学习

EasyExcel是alibaba公司开源的优秀框架，有优秀的开发人员在持续维护。可以通过EasyExcel的官方文档来学习使用，也可以下载项目源码来了解实现原理。

1.3 使用EaseExcel

项目中只需要引入EaseExcel的依赖信息，就可以编码使用Jar包中定义的类方法实现Excel文件的导入和导出。

1
2
3
4
5
6
xml复制代码<!-- EaseExcel的依赖信息 -->
<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>easyexcel</artifactId>
    <version>3.0.5</version>
</dependency>
  1. EasyExcel的导出实现

导出功能相比于导入有着更加广泛的使用，因此首先来学习EasyExcel中导出Excel的实现。

2.1 建立模型类

Java语言中，所有的数据是基于对象存在，而数据导出时也脱离不数据对象，因此首先要建立与导出数据结构相同的模型类。

定义一个简单的Java类来承载用户数据，使用lombok的@Data快速实现对象创建。

1
2
3
4
5
6
7
java复制代码//创建数据模型，存放用户导出数据
@Data
public class UserModel {
    private String id;
    private String code;
    private String name;
}

2.2 本地导出方法实现

本地简单实现Excel文件的导出时，只需要调用EasyExcel类中的相关方法定义导出文件的属性，并使用doWrite()设置导出数据后完成Excel的导出。

1
2
3
4
5
6
7
8
9
10
11
12
java复制代码@Test
public void exportTemplate(){
    List<UserModel> list = new ArrayList<>();
    UserModel userModel = new UserModel();
    userModel.setCode("123");
    userModel.setName("tom");
    list.add(userModel);

    EasyExcel.write("用户数据.xlsx", UserModel.class)
            .sheet("Sheet1")
            .doWrite(list);
}

对于导出代码中的类方法，可以简单介绍为：

  • EasyExcel作为工具类，继承EasyExcelFactory，以建造者模式设置Excel的属性，最终完成创建和导出
  • write()方法定义了文件导出路径信息pathName、以及使用哪个类class来写数据
  • sheet()方法可以对导出Excel的sheet指定名称
  • doWrite()用于指定Excel中的数据
  • 文件导出结束后文件流会自动关闭
  1. EasyExcel导入的实现

Excel导入时就是对选择文件内容的读取操作，使用EasyExcel读取文件时的流程如下：

  1. 创建一个实体对象用来承载从文件中读取的数据，对象结构要和文件中数据一致
  2. 定义一个回调监听器，用于在一行一行读取数据时执行回调
  3. 通过EasyExcel读取文件

3.1 创建实体对象

创建方式与导出数据时一致，只要保证对象的属性与导入数据列一致即可。

1
2
3
4
5
6
java复制代码//导入时不需要额外的列，导入导出也可以使用同一个实体类
@Data
public class UserModel {
    private String code;
    private String name;
}

3.2 定义监听器

EasyExcel要求定义一个监听器实现ReadListener类，其中T代表当前要导入的实体类，并实现其中的invoke()和doAfterAllAnalysed()两个方法。

  • invoke()方法用于每次读取一行数据后回调
  • doAfterAllAnalysed()方法用于整个文件读取完成后回调
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
java复制代码//导入时的监听器要针对每个导入对象类创建
public class UserModelListener implements ReadListener<UserModel> {
    private static final Logger log  = LoggerFactory.getLogger(UserModelListener.class);
    @Override
    public void invoke(UserModel userModel, AnalysisContext analysisContext) {
        log.info("解析到一条数据:{}", userModel.toString());
        //指定其他操作
    }

    @Override
    public void doAfterAllAnalysed(AnalysisContext analysisContext) {
        log.info("所有数据解析完成！");
        //指定其他操作
    }
}

3.3 本地导入方法实现

定义好实体类和对应的回调监听器后，便可以使用EasyExcel来实现Excel文件的读取导入。

  • read()方法用来指定读取文件的路径信息、读取实体类、回调监听器类等信息
  • sheet()方法用来指定读取文件中的sheet名称
  • doRead()方法则是执行读取操作，每读取一行数据便会执行一次回调
1
2
3
4
5
6
7
java复制代码//基于指定路径文件、实体类、实体类的监听器实现导入操作
@Test
public void importExcel(){
    EasyExcel.read("用户数据.xlsx", UserModel.class, new UserModelListener())
            .sheet()
            .doRead();
}

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%