public void addProducer ( Broker broker ) { Properties props = new Properties ( ) ; props . put ( "host" , broker . host ) ; props . put ( "port" , "" + broker . port ) ; props . putAll ( config . getProperties ( ) ) ; if ( sync ) { SyncProducer producer = new SyncProducer ( new SyncProducerConfig ( props ) ) ; logger . info ( "Creating sync producer for broker id = " + broker . id + " at " + broker . host + ":" + broker . port ) ; syncProducers . put ( broker . id , producer ) ; } else { AsyncProducer < V > producer = new AsyncProducer < V > ( new AsyncProducerConfig ( props ) , // new SyncProducer ( new SyncProducerConfig ( props ) ) , // serializer , // eventHandler , // config . getEventHandlerProperties ( ) , // this . callbackHandler , // config . getCbkHandlerProperties ( ) ) ; producer . start ( ) ; logger . info ( "Creating async producer for broker id = " + broker . id + " at " + broker . host + ":" + broker . port ) ; asyncProducers . put ( broker . id , producer ) ; } }
add a new producer either synchronous or asynchronous connecting to the specified broker