consumer 才用 pull 的方式 从 broker 中读数据