This repository has been archived by the owner on Aug 20, 2021. It is now read-only.
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
新建 ThreadUtil.execute(List<T>, PartitionThreadConfig, Map<String, ?>,
PartitionPerHandler<T>) fix #803 新建 ThreadUtil.execute(List<T>, int, Map<String, ?>, PartitionPerHandler<T>) fix #802 新建 ThreadUtil.execute(List<T>, int, PartitionPerHandler<T>) fix #801
- Loading branch information
1 parent
bf45176
commit 6d184c4
Showing
17 changed files
with
682 additions
and
31 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
34 changes: 34 additions & 0 deletions
34
src/main/java/com/feilong/core/lang/thread/PartitionEachSizeBuilder.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,34 @@ | ||
| /* | ||
| * Copyright (C) 2008 feilong | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package com.feilong.core.lang.thread; | ||
|
|
||
| /** | ||
| * 用来计算 each size 大小的. | ||
| * | ||
| * @author <a href="http://feitianbenyue.iteye.com/">feilong</a> | ||
| * @since 2.0.0 | ||
| */ | ||
| public interface PartitionEachSizeBuilder{ | ||
|
|
||
| /** | ||
| * Builds the. | ||
| * | ||
| * @param totalSize | ||
| * the total size | ||
| * @return 如果 <code>totalSize<=0</code> 是empty,抛出 {@link IllegalArgumentException}<br> | ||
| */ | ||
| int build(int totalSize); | ||
| } |
115 changes: 115 additions & 0 deletions
115
src/main/java/com/feilong/core/lang/thread/PartitionEachSizeThreadConfigBuilder.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,115 @@ | ||
| /* | ||
| * Copyright (C) 2008 feilong | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package com.feilong.core.lang.thread; | ||
|
|
||
| import static com.feilong.core.bean.ConvertUtil.toInteger; | ||
|
|
||
| import java.math.RoundingMode; | ||
|
|
||
| import org.apache.commons.lang3.Validate; | ||
|
|
||
| import com.feilong.core.lang.NumberUtil; | ||
|
|
||
| /** | ||
| * 用来计算 each size 大小的. | ||
| * | ||
| * @author <a href="http://feitianbenyue.iteye.com/">feilong</a> | ||
| * @since 2.0.0 | ||
| */ | ||
| public class PartitionEachSizeThreadConfigBuilder implements PartitionEachSizeBuilder{ | ||
|
|
||
| /** The partition thread config. */ | ||
| private PartitionThreadConfig partitionThreadConfig; | ||
|
|
||
| //--------------------------------------------------------------- | ||
|
|
||
| /** | ||
| * Instantiates a new partition each size thread config builder. | ||
| */ | ||
| public PartitionEachSizeThreadConfigBuilder(){ | ||
| super(); | ||
| } | ||
|
|
||
| /** | ||
| * Instantiates a new partition each size thread config builder. | ||
| * | ||
| * @param partitionThreadConfig | ||
| * the partition thread config | ||
| */ | ||
| public PartitionEachSizeThreadConfigBuilder(PartitionThreadConfig partitionThreadConfig){ | ||
| super(); | ||
| this.partitionThreadConfig = partitionThreadConfig; | ||
| } | ||
|
|
||
| //--------------------------------------------------------------- | ||
| /** | ||
| * 构造每个分区大小. | ||
| * | ||
| * @param totalSize | ||
| * the total size | ||
| * @return 如果 <code>totalSize<=0</code> 是empty,抛出 {@link IllegalArgumentException}<br> | ||
| * 如果 <code>partitionConfig</code> 是null,抛出 {@link NullPointerException}<br> | ||
| * 如果 <code>partitionConfig.maxThreadCount<=0</code> 是empty,抛出 {@link IllegalArgumentException}<br> | ||
| * 如果 <code>partitionConfig.minPerThreadHandlerCount<=0</code> 是empty,抛出 {@link IllegalArgumentException}<br> | ||
| * | ||
| * 如果 <code>totalSize</code> 小于等于 <code>minPerThreadHandlerCount</code>(每个线程最少处理数量),那么直接返回<code>totalSize</code>,也就是说接下来开 1 | ||
| * 个线程就足够了<br> | ||
| */ | ||
| @Override | ||
| public int build(int totalSize){ | ||
| Validate.isTrue(totalSize > 0, "totalSize must >0,totalSize:%s", totalSize); | ||
| Validate.notNull(partitionThreadConfig, "partitionConfig can't be null!"); | ||
| //--------------------------------------------------------------- | ||
| //启动最大线程数 | ||
| int maxThreadCount = partitionThreadConfig.getMaxThreadCount(); | ||
| // 每个线程最少处理数量 | ||
| int minPerThreadHandlerCount = partitionThreadConfig.getMinPerThreadHandlerCount(); | ||
|
|
||
| Validate.isTrue(maxThreadCount > 0, "maxThreadCount must >0,totalSize:%s", maxThreadCount); | ||
| Validate.isTrue(minPerThreadHandlerCount > 0, "minPerThreadHandlerCount must >0,totalSize:%s", minPerThreadHandlerCount); | ||
| //--------------------------------------------------------------- | ||
| //如果 totalSize 小于等于 minPerThreadHandlerCount(每个线程最少处理数量), 那么直接返回totalSize ,也就是说接下来开 1 个线程就足够了 | ||
| if (totalSize <= minPerThreadHandlerCount){ | ||
| return totalSize; | ||
| } | ||
|
|
||
| //--------------------------------------------------------------- | ||
| //用总数 totalSize 除以 minPerThreadHandlerCount (每个线程最少处理数量),向上取整 | ||
| int threadCount = toInteger(NumberUtil.getDivideValue(totalSize, minPerThreadHandlerCount, 0, RoundingMode.UP)); | ||
| //如果 算出来的线程数 要超过 maxThreadCount,比如 总数 100 条, 但是设置的minPerThreadHandlerCount (每个线程最少处理数量) 是 20, 最大线程数是 4 ,此时 100/20 是 5 ,大约最大线程数 4 | ||
|
|
||
| //那么最大只能开启 4 个线程, 返回的每个线程处理的数量是 100/4 =25 | ||
| if (threadCount >= maxThreadCount){ | ||
| return toInteger(NumberUtil.getDivideValue(totalSize, maxThreadCount, 0, RoundingMode.UP)); | ||
| } | ||
|
|
||
| //否则返回 threadCount | ||
| return threadCount; | ||
| } | ||
|
|
||
| //--------------------------------------------------------------- | ||
|
|
||
| /** | ||
| * Sets the partition thread config. | ||
| * | ||
| * @param partitionThreadConfig | ||
| * the partitionThreadConfig to set | ||
| */ | ||
| public void setPartitionThreadConfig(PartitionThreadConfig partitionThreadConfig){ | ||
| this.partitionThreadConfig = partitionThreadConfig; | ||
| } | ||
|
|
||
| } |
126 changes: 126 additions & 0 deletions
126
src/main/java/com/feilong/core/lang/thread/PartitionThreadConfig.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,126 @@ | ||
| /* | ||
| * Copyright (C) 2008 feilong | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package com.feilong.core.lang.thread; | ||
|
|
||
| import org.apache.commons.lang3.builder.ToStringBuilder; | ||
| import org.apache.commons.lang3.builder.ToStringStyle; | ||
|
|
||
| /** | ||
| * 分区控制. | ||
| * | ||
| * @author <a href="http://feitianbenyue.iteye.com/">feilong</a> | ||
| * @since 2.0.0 | ||
| */ | ||
| public class PartitionThreadConfig{ | ||
|
|
||
| /** Static instance. */ | ||
| // the static instance works for all types | ||
| public static final PartitionThreadConfig INSTANCE = new PartitionThreadConfig(); | ||
|
|
||
| //--------------------------------------------------------------- | ||
|
|
||
| /** 启动最大线程数. */ | ||
| private int maxThreadCount = 100; | ||
|
|
||
| /** 每个线程最少处理数量. */ | ||
| private int minPerThreadHandlerCount = 20; | ||
|
|
||
| //--------------------------------------------------------------- | ||
|
|
||
| /** | ||
| * Instantiates a new partition each size entity. | ||
| */ | ||
| public PartitionThreadConfig(){ | ||
| super(); | ||
| } | ||
|
|
||
| /** | ||
| * Instantiates a new partition each size entity. | ||
| * | ||
| * @param minPerThreadHandlerCount | ||
| * 每个线程最少处理数量 | ||
| */ | ||
| public PartitionThreadConfig(int minPerThreadHandlerCount){ | ||
| super(); | ||
| this.minPerThreadHandlerCount = minPerThreadHandlerCount; | ||
| } | ||
|
|
||
| /** | ||
| * Instantiates a new partition each size entity. | ||
| * | ||
| * @param maxThreadCount | ||
| * 启动最大线程数 | ||
| * @param minPerThreadHandlerCount | ||
| * 每个线程最少处理数量 | ||
| */ | ||
| public PartitionThreadConfig(int maxThreadCount, int minPerThreadHandlerCount){ | ||
| super(); | ||
| this.maxThreadCount = maxThreadCount; | ||
| this.minPerThreadHandlerCount = minPerThreadHandlerCount; | ||
| } | ||
|
|
||
| //--------------------------------------------------------------- | ||
|
|
||
| /** | ||
| * 获得 启动最大线程数. | ||
| * | ||
| * @return the maxThreadCount | ||
| */ | ||
| public int getMaxThreadCount(){ | ||
| return maxThreadCount; | ||
| } | ||
|
|
||
| /** | ||
| * 设置 启动最大线程数. | ||
| * | ||
| * @param maxThreadCount | ||
| * the maxThreadCount to set | ||
| */ | ||
| public void setMaxThreadCount(int maxThreadCount){ | ||
| this.maxThreadCount = maxThreadCount; | ||
| } | ||
|
|
||
| /** | ||
| * 获得 每个线程最少处理数量. | ||
| * | ||
| * @return the minPerThreadHandlerCount | ||
| */ | ||
| public int getMinPerThreadHandlerCount(){ | ||
| return minPerThreadHandlerCount; | ||
| } | ||
|
|
||
| /** | ||
| * 设置 每个线程最少处理数量. | ||
| * | ||
| * @param minPerThreadHandlerCount | ||
| * the minPerThreadHandlerCount to set | ||
| */ | ||
| public void setMinPerThreadHandlerCount(int minPerThreadHandlerCount){ | ||
| this.minPerThreadHandlerCount = minPerThreadHandlerCount; | ||
| } | ||
|
|
||
| //--------------------------------------------------------------- | ||
| /* | ||
| * (non-Javadoc) | ||
| * | ||
| * @see java.lang.Object#toString() | ||
| */ | ||
| @Override | ||
| public String toString(){ | ||
| return ToStringBuilder.reflectionToString(this, ToStringStyle.SHORT_PREFIX_STYLE); | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.