下载配置kafka参考该链接：http://www.cnblogs.com/super-d2/p/4534323.html

pom.xml:

                 
      
       org.apache.kafka
                  
      
       kafka_2.10
                  
      
       0.8.2.1
              
             
                 
      
       org.springframework.integration
                  
      
       spring-integration-kafka
                  
      
       1.2.0.RELEASE
              
     

 

producer配置文件：

     
         
          
          
          
          
              
                   
                        
         
          3600000
                         
         
          5
                         
         
          5242880
                     
                
           
          
          
              
                   
                
           
      
     

consumer配置：

      
          
           
           
               
            
           
           
           
               
                    
                         
         
          smallest
                          
         
          10485760
          
                          
         
          5242880
                          
         
          1000
                      
                 
            
           
           
           
           
               
                    
                         
                      
                 
            
       
     

producer 测试代码：

public class KafkaServiceImpl implements KafkaService {    @Autowired    @Qualifier("inputToKafka")    MessageChannel channel;    public void sendUserInfo(String key, Object obj) {        Message msg = MessageBuilder.withPayload(obj)                .setHeader("kafkaUser", key)                .setHeader(KafkaHeaders.TOPIC, "helloworldTopic").build();        channel.send(msg);    }}

consumer测试代码：

package cn.innmall.union.function.service.work;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import java.util.Map;/** * Created by yujinghui on 4/23/16. * */public class KafkaService {    static final Logger logger = LoggerFactory.getLogger(KafkaService.class);    public void processMessage(Map
     
      > msgs) {        for (Map.Entry < String,Map
      
       >entry:        msgs.entrySet()){            System.out.println("Consumer Message received: ");            logger.debug("Suchit Topic:" + entry.getKey());            for (String msg : entry.getValue().values()) {                logger.info("Suchit Consumed Message: " + msg);            }        }    }}