@Ignore @Test public void getConnTest() throws Exception { PropUtils p = PropUtils.getInstance(); int kafkaConnPoolSize = Integer.parseInt(p.getProperty("kafka.conn.pool.size")); Producer<String, byte[]> conn = KafkaConnPoolUtils.getInstance().getConn(); String topic = p.getProperty("kafka.topic"); for (int i = 0, max = 1000000000; i < max; i++) { System.out.println(String.format("Sending %s/%s ...", i, max)); Thread.sleep(1000); conn.send(new KeyedMessage<>(topic, ("{\"appId\":1,\"attemptId\":\"2\",\"callId\":\"" + i + "\",\"description\":\"yuzhouwan\"}") .getBytes())); } for (int i = 1; i < 2 * kafkaConnPoolSize; i++) KafkaConnPoolUtils.getInstance().getConn(); }
public Producer<String, byte[]> getConn() { long index = (CONN_INDEX %= CONN_IN_POOL); CONN_INDEX++; _log.debug("Get Kafka Connection from pool, index: [{} in {}] ...", index + 1, CONN_IN_POOL); return pool.get(index + ""); }
KafkaConnPoolUtils { public Producer<String, byte[]> getConn() { long index = (CONN_INDEX %= CONN_IN_POOL); CONN_INDEX++; _log.debug("Get Kafka Connection from pool, index: [{} in {}] ...", index + 1, CONN_IN_POOL); return pool.get(index + ""); } }
KafkaConnPoolUtils { public Producer<String, byte[]> getConn() { long index = (CONN_INDEX %= CONN_IN_POOL); CONN_INDEX++; _log.debug("Get Kafka Connection from pool, index: [{} in {}] ...", index + 1, CONN_IN_POOL); return pool.get(index + ""); } private  KafkaConnPoolUtils(); }
KafkaConnPoolUtils { public Producer<String, byte[]> getConn() { long index = (CONN_INDEX %= CONN_IN_POOL); CONN_INDEX++; _log.debug("Get Kafka Connection from pool, index: [{} in {}] ...", index + 1, CONN_IN_POOL); return pool.get(index + ""); } private  KafkaConnPoolUtils(); static KafkaConnPoolUtils getInstance(); static Collection<Producer<String, byte[]>> getPool(); Producer<String, byte[]> getConn(); }
KafkaConnPoolUtils { public Producer<String, byte[]> getConn() { long index = (CONN_INDEX %= CONN_IN_POOL); CONN_INDEX++; _log.debug("Get Kafka Connection from pool, index: [{} in {}] ...", index + 1, CONN_IN_POOL); return pool.get(index + ""); } private  KafkaConnPoolUtils(); static KafkaConnPoolUtils getInstance(); static Collection<Producer<String, byte[]>> getPool(); Producer<String, byte[]> getConn();  }