使用自定义CONNECTION FACTORY，这样会覆盖SPRING 的AUTO CONFIGURATION。
ActiveMQConnectionFactoryFactory.java
import java.lang.reflect.InvocationTargetException;
import java.util.Collections;
import java.util.List;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.springframework.boot.autoconfigure.jms.activemq.ActiveMQConnectionFactoryCustomizer;
import org.springframework.boot.autoconfigure.jms.activemq.ActiveMQProperties;
import org.springframework.boot.autoconfigure.jms.activemq.ActiveMQProperties.Packages;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* Factory to create a {@link ActiveMQConnectionFactory} instance from properties defined
* in {@link SecondaryActiveMQProperties}.
*
* @author Phillip Webb
* @author Venil Noronha
*/class ActiveMQConnectionFactoryFactory {
private static final String DEFAULT_EMBEDDED_BROKER_URL = "vm://localhost?broker.persistent=false";
private static final String DEFAULT_NETWORK_BROKER_URL = "tcp://localhost:61616";
private final ActiveMQProperties properties;
private final List<ActiveMQConnectionFactoryCustomizer> factoryCustomizers;
ActiveMQConnectionFactoryFactory(ActiveMQProperties properties,
List<ActiveMQConnectionFactoryCustomizer> factoryCustomizers) {
Assert.notNull(properties, "Properties must not be null");
this.properties = properties;
this.factoryCustomizers = (factoryCustomizers !=
null) ? factoryCustomizers : Collections.emptyList();
}
public <T
extends ActiveMQConnectionFactory> T createConnectionFactory(Class<T> factoryClass) {
try {
return doCreateConnectionFactory(factoryClass);
}
catch (Exception ex) {
throw new IllegalStateException("Unable to create " + "ActiveMQConnectionFactory", ex);
}
}
private <T
extends ActiveMQConnectionFactory> T doCreateConnectionFactory(Class<T> factoryClass)
throws Exception {
T factory = createConnectionFactoryInstance(factoryClass);
if (
this.properties.getCloseTimeout() !=
null) {
factory.setCloseTimeout((
int)
this.properties.getCloseTimeout().toMillis());
}
factory.setNonBlockingRedelivery(
this.properties.isNonBlockingRedelivery());
if (
this.properties.getSendTimeout() !=
null) {
factory.setSendTimeout((
int)
this.properties.getSendTimeout().toMillis());
}
Packages packages =
this.properties.getPackages();
if (packages.getTrustAll() !=
null) {
factory.setTrustAllPackages(packages.getTrustAll());
}
if (!packages.getTrusted().isEmpty()) {
factory.setTrustedPackages(packages.getTrusted());
}
customize(factory);
return factory;
}
private <T
extends ActiveMQConnectionFactory> T createConnectionFactoryInstance(Class<T> factoryClass)
throws InstantiationException, IllegalAccessException, InvocationTargetException, NoSuchMethodException {
String brokerUrl = determineBrokerUrl();
String user =
this.properties.getUser();
String password =
this.properties.getPassword();
if (StringUtils.hasLength(user) && StringUtils.hasLength(password)) {
return factoryClass.getConstructor(String.
class, String.
class, String.
class).newInstance(user, password,
brokerUrl);
}
return factoryClass.getConstructor(String.
class).newInstance(brokerUrl);
}
private void customize(ActiveMQConnectionFactory connectionFactory) {
for (ActiveMQConnectionFactoryCustomizer factoryCustomizer :
this.factoryCustomizers) {
factoryCustomizer.customize(connectionFactory);
}
}
String determineBrokerUrl() {
if (
this.properties.getBrokerUrl() !=
null) {
return this.properties.getBrokerUrl();
}
if (
this.properties.isInMemory()) {
return DEFAULT_EMBEDDED_BROKER_URL;
}
return DEFAULT_NETWORK_BROKER_URL;
}
}
TwinJmsConnectionFactoryConfiguration.java
import java.util.stream.Collectors;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.messaginghub.pooled.jms.JmsPoolConnectionFactory;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.jms.JmsPoolConnectionFactoryFactory;
import org.springframework.boot.autoconfigure.jms.activemq.ActiveMQConnectionFactoryCustomizer;
import org.springframework.boot.autoconfigure.jms.activemq.ActiveMQProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.context.annotation.Profile;
@Configuration
@Profile({"local"})
public class TwinJmsConnectionFactoryConfiguration {
@Bean
@ConfigurationProperties(prefix = "spring.activemq.primary")
public ActiveMQProperties primaryActiveMQProperties() {
return new ActiveMQProperties();
}
@Bean(destroyMethod = "stop")
@Primary
@ConditionalOnProperty(prefix = "spring.activemq.pool", name = "enabled", havingValue = "true")
public JmsPoolConnectionFactory connectionFactory(ActiveMQProperties primaryActiveMQProperties,
ObjectProvider<ActiveMQConnectionFactoryCustomizer> factoryCustomizers) {
ActiveMQConnectionFactory connectionFactory =
new ActiveMQConnectionFactoryFactory(primaryActiveMQProperties,
factoryCustomizers.orderedStream().collect(Collectors.toList()))
.createConnectionFactory(ActiveMQConnectionFactory.
class);
return new JmsPoolConnectionFactoryFactory(primaryActiveMQProperties.getPool())
.createPooledConnectionFactory(connectionFactory);
}
////////////////////////////////////////////////////////////////////////////////
@Bean
@ConfigurationProperties(prefix = "spring.activemq.sescond")
public ActiveMQProperties sescondActiveMQProperties() {
return new ActiveMQProperties();
}
@Bean(destroyMethod = "stop")
@ConditionalOnProperty(prefix = "spring.activemq.pool", name = "enabled", havingValue = "true")
public JmsPoolConnectionFactory sescondPooledJmsConnectionFactory(ActiveMQProperties sescondActiveMQProperties,
ObjectProvider<ActiveMQConnectionFactoryCustomizer> factoryCustomizers) {
ActiveMQConnectionFactory connectionFactory =
new ActiveMQConnectionFactoryFactory(sescondActiveMQProperties,
factoryCustomizers.orderedStream().collect(Collectors.toList()))
.createConnectionFactory(ActiveMQConnectionFactory.
class);
return new JmsPoolConnectionFactoryFactory(sescondActiveMQProperties.getPool())
.createPooledConnectionFactory(connectionFactory);
}
}
这时发现如果处理总数足够大时，被处理的ITEMS总数会少于应该处理的总数。
+------------+--------------+-------------+-----------------+------------------+--------------------+----------------+-----------+-------------------------
| READ_COUNT | FILTER_COUNT | WRITE_COUNT | READ_SKIP_COUNT | WRITE_SKIP_COUNT | PROCESS_SKIP_COUNT | ROLLBACK_COUNT | EXIT_CODE | EXIT_MESSAGE
-+------------+--------------+-------------+-----------------+------------------+--------------------+----------------+-----------+-------------------------
| 1 | 0 | 1 | 0 | 0 | 0 | 0 | COMPLETED |
| 30006 | 0 | 29947 | 0 | 59 | 0 | 61 | COMPLETED | Waited for 101 results.
To see how many connections are configured for your DB to use:
select @@max_connections;
To change it:
set global max_connections = 200;
To see how many are connected at the current time:
show processlist;
vi /etc/my.cnf
[mysqld]
max_connections = 500
MongDB Client请求查询数据，需要包括五个阶段：
MongoDB Client需要找到可用的MongoDB Server
MongoDB Client需要和MongoDB Server建立（new）Connection
应用程序处理线程从Connection Pool中获取Connection
数据传输（获取连接后，进行Socket通信，获取数据）
断开Collection
那么，MongoDB Client驱动设置中网络相关等待超时参数serverSelectionTimeout、connectTimeout、maxWaitTime和socketTimeout分别对应上面哪个环节呢？
参数serverSelectionTimeout：对应第1个环节，即MongoDB Client需要找到可用的MongoDB Server所需要的等待时间， MongDB部署的生产一般由多个服务器组成，要么作为一个复制集或者作为一个分片集群，参数 serverSelectionTimeout的值即为多长时间内找不到合适服务器时候就决定放弃的时间间隔；
参数connectTimeout：对应第2个环节，每次创建Connection，对应的网络等待。单位毫秒数, 0表示没有限制；
参数maxWaitTime：对应第3个环节，应用程序处理线程从连接池中获取Collection，对应的网络等待时间。单位毫秒数，0表示 不等待，负数表示等待时间不确定；
参数socketTimeout：对应第4个环节，获取Connection后，就有了Socket通信，获取数据，即在MonogoDB Client 和MonogoDB Server的Socket通信过程中的网络等待时间。单位毫秒数，默认配置为0，也就是没有限制， 没有超 时限制，系统出了问题也不容易发现，应该根据实际情况，给出合理的超时时间。
其他相关参数如下：
connectionsPerHost：对mongo实例来说，每个host允许链接的最大链接数,这些链接空闲时会放入池中,如果链接被耗尽，任何请求链接的操作会被阻塞等待链接可用,推荐配置10
minPoolsSize：当Connection空闲时,Connection Pool中最少Connection保有量；
threadsAllowedToBlockForConnectionMultiplier：每个Connection的可以阻塞等待的线程队列数，它以上面connectionsPerHost值相乘的结果就是阻塞等待的线程队列最大值。如果连接线程排满了队列就会抛出“Out of semaphores to get db”错误。
socketKeepAlive:该标志用于控制socket保持活动的功能，通过防火墙保持连接活着
socketKeepAlive=false
autoConnectRetry：这个控制是否在一个Connection时，系统会自动重试
#true:假如Connection不能建立时,驱动将重试相同的server,有最大的重试次数,默认为15次,这样可以避免一些server因为一些阻塞操作零时down而驱动抛出异常,这个对平滑过度到一个新的master,也是很有用的,注意:当集群为复制集时,驱动将在这段时间里,尝试链接到旧的master上,而不会马上链接到新master上
#false 当在进行socket读写时,不会阻止异常抛出,驱动已经有自动重建破坏链接和重试读操作. 推荐配置false
autoConnectRetry=false
#重新打开链接到相同server的最大毫秒数,推荐配置为0，如果 autoConnectRetry=true,表示时间为15s
#com.jd.mongodbclient2.mongo.JDClientMongo.maxAutoConnectRetryTime=false
————————————————
版权声明：本文为CSDN博主「pursuer211」的原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/pursuer211/article/details/82994027