何时使用MQ
谈到Kafka
就不得不提到MQ,是属于消息队列的一种。作为一种基础中间件在互联网项目中有着大量的使用。
一种技术的产生自然是为了解决某种需求,通常来说是以下场景:
- 需要跨进程通信:B系统需要A系统的输出作为输入参数。
- 当A系统的输出能力远远大于B系统的处理能力。
针对于第一种情况有两种方案:
- 使用
RPC
远程调用,A直接调用B。
- 使用
MQ
,A发布消息到MQ
,B订阅该消息。
当我们的需求是:A调用B实时响应,并且实时关心响应结果则使用RPC
,这种情况就得使用同步调用。
反之当我们并不关心调用之后的执行结果,并且有可能被调用方的执行非常耗时,这种情况就非常适合用MQ
来达到异步调用目的。
比如常见的登录场景就只能用同步调用的方式,因为这个过程需要实时的响应结果,总不能在用户点了登录之后排除网络原因之外再额外的等几秒吧。
但类似于用户登录需要奖励积分的情况则使用MQ
会更好,因为登录并不关系积分的情况,只需要发个消息到MQ
,处理积分的服务订阅处理即可,这样还可以解决积分系统故障带来的雪崩效应。
MQ
还有一个基础功能则是限流削峰,这对于大流量的场景如果将请求直接调用到B系统则非常有可能使B系统出现不可用的情况。这种场景就非常适合将请求放入MQ
,不但可以利用MQ
削峰还尽可能的保证系统的高可用。
Kafka简介
本次重点讨论下Kafka
。
简单来说Kafka
是一个支持水平扩展,高吞吐率的分布式消息系统。
Kafka
的常用知识:
Topic
:生产者和消费者的交互都是围绕着一个Topic
进行的,通常来说是由业务来进行区分,由生产消费者协商之后进行创建。
Partition
(分区):是Topic
下的组成,通常一个Topic
下有一个或多个分区,消息生产之后会按照一定的算法负载到每个分区,所以分区也是Kafka
性能的关键。当发现性能不高时便可考虑新增分区。
结构图如下:
创建Topic
Kafka
的安装官网有非常详细的讲解。这里谈一下在日常开发中常见的一些操作,比如创建Topic
:
sh bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic `test`
创建了三个分区的test
主题。
使用
sh bin/kafka-topics.sh --list --zookeeper localhost:2181
可以列出所有的Topic
。
Kafka生产者
使用kafka
官方所提供的Java API
来进行消息生产,实际使用中编码实现更为常用:
/** Kafka生产者 * @author crossoverJie */ public class Producer { private static final Logger LOGGER = LoggerFactory.getLogger(Producer.class); /** * 消费配置文件 */ private static String consumerProPath; public static void main(String[] args) throws IOException { // set up the producer consumerProPath = System.getProperty("product_path"); KafkaProducer<String, String> producer = null; try { FileInputStream inputStream = new FileInputStream(new File(consumerProPath)); Properties properties = new Properties(); properties.load(inputStream); producer = new KafkaProducer<String, String>(properties); } catch (IOException e) { LOGGER.error("load config error", e); } try { // send lots of messages for (int i=0 ;i<100 ; i++){ producer.send(new ProducerRecord<String, String>( "topic_optimization", i+"", i+"")); } } catch (Throwable throwable) { System.out.printf("%s", throwable.getStackTrace()); } finally { producer.close(); } } }
再配合以下启动参数即可发送消息:
-Dproduct_path=/xxx/producer.properties