登录 注册

深入学习 RabbitMQ(三)：channel 的 confirm 模式

本贴最后更新于 2282 天前，其中的信息可能已经时移俗易

[转]http://blog.csdn.net/hzw19920329/article/details/54340711

上一篇博客我们介绍了使用 RabbitMQ 可能会遇到的一个问题，即生产者不知道消息是否真正到达 broker 代理服务器，随后通过 AMQP 协议层面为我们提供的事务机制解决了这个问题，但是采用事务机制实现会降低 RabbitMQ 的消息吞吐量，那么有没有更加高效的解决方式呢？RabbitMQ 团队为我们拿出了更好的方案，即采用发送方确认模式；

   生产者确认模式实现原理：

   生产者将信道设置成confirm模式，一旦信道进入confirm模式，所有在该信道上面发布的消息都将会被指派一个唯一的ID(从1开始)，一旦消息被投递到所有匹配的队列之后，broker就会发送一个确认给生产者(包含消息的唯一ID)，这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出，broker回传给生产者的确认消息中delivery-tag域包含了确认消息的序列号，此外broker也可以设置basic.ack的multiple域，表示到这个序列号之前的所有消息都已经得到了处理；

   confirm模式最大的好处在于他是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果RabbitMQ因为自身内部错误导致消息丢失，就会发送一条nack消息，生产者应用程序同样可以在回调方法中处理该nack消息；

   开启confirm模式的方法：

   生产者通过调用channel的confirmSelect方法将channel设置为confirm模式，(注意一点，已经在transaction事务模式的channel是不能再设置成confirm模式的，即这两种模式是不能共存的)，如果没有设置no-wait标志的话，broker会返回confirm.select-ok表示同意发送者将当前channel信道设置为confirm模式(从目前RabbitMQ最新版本3.6来看，如果调用了channel.confirmSelect方法，默认情况下是直接将no-wait设置成false的，也就是默认情况下broker是必须回传confirm.select-ok的，而且我也没找到我们自己能够设置no-wait标志的方法)；

   生产者实现confiem模式有三种编程方式：

   (1)：普通confirm模式，每发送一条消息，调用waitForConfirms()方法等待服务端confirm，这实际上是一种串行的confirm，每publish一条消息之后就等待服务端confirm，如果服务端返回false或者超时时间内未返回，客户端进行消息重传；

   (2)：批量confirm模式，每发送一批消息之后，调用waitForConfirms()方法，等待服务端confirm，这种批量确认的模式极大的提高了confirm效率，但是如果一旦出现confirm返回false或者超时的情况，客户端需要将这一批次的消息全部重发，这会带来明显的重复消息，如果这种情况频繁发生的话，效率也会不升反降；

   

   讲完了基本的原理之后，代码级别我们该怎么设置channel信道为confirm模式呢？以及我们该怎么获取broker返回给我们的确认消息呢？

   测试1：普通confirm模式

   首先从最简单的开始，仅仅将channel设置成confirm模式，并且生产者每发送一条消息就等待broker回应确认消息，至于确认消息是什么我们不去做任何处理，为了测试方便，此处生产者只发送了5条消息，实现代码如下：

[java] view plain copy

  1. public class ProducerTest {

  2.  public static void main(String[] args) {  

  3.      String exchangeName = "confirmExchange";  

  4.      String queueName = "confirmQueue";  

  5.      String routingKey = "confirmRoutingKey";  

  6.      String bindingKey = "confirmRoutingKey";  

  7.      int count = 5;  

  9.      ConnectionFactory factory = new ConnectionFactory();  

  10.      factory.setHost("172.16.151.74");  

  11.      factory.setUsername("test");  

  12.      factory.setPassword("test");  

  13.      factory.setPort(5672);  

  15.      //创建生产者  

  16.      Sender producer = new Sender(factory, count, exchangeName, queueName,routingKey,bindingKey);  

  17.      producer.run();  

  18.  }

  19. }

  21. class Sender

  22. {

  23.  private ConnectionFactory factory;  

  24.  private int count;  

  25.  private String exchangeName;  

  26.  private String  queueName;  

  27.  private String routingKey;  

  28.  private String bindingKey;  

  30.  public Sender(ConnectionFactory factory,int count,String exchangeName,String queueName,String routingKey,String bindingKey) {  

  31.      this.factory = factory;  

  32.      this.count = count;  

  33.      this.exchangeName = exchangeName;  

  34.      this.queueName = queueName;  

  35.      this.routingKey = routingKey;  

  36.      this.bindingKey = bindingKey;  

  37.  }  

  39.  public void run() {  

  40.      Channel channel = null;  

  41.      try {  

  42.          Connection connection = factory.newConnection();  

  43.          channel = connection.createChannel();  

  44.          //创建exchange  

  45.          channel.exchangeDeclare(exchangeName, "direct", true, false, null);  

  46.          //创建队列  

  47.          channel.queueDeclare(queueName, true, false, false, null);  

  48.          //绑定exchange和queue  

  49.          channel.queueBind(queueName, exchangeName, bindingKey);  

  50.          channel.confirmSelect();  

  51.          //发送持久化消息  

  52.          for(int i = 0;i < count;i++)  

  53.          {  

  54.              //第一个参数是exchangeName(默认情况下代理服务器端是存在一个""名字的exchange的,  

  55.              //因此如果不创建exchange的话我们可以直接将该参数设置成"",如果创建了exchange的话  

  56.              //我们需要将该参数设置成创建的exchange的名字),第二个参数是路由键  

  57.              channel.basicPublish(exchangeName, routingKey,MessageProperties.PERSISTENT_BASIC, ("第"+(i+1)+"条消息").getBytes());  

  58.              if(channel.waitForConfirms())  

  59.              {  

  60.                  System.out.println("发送成功");  

  61.              }  

  62.          }  

  63.          final long start = System.currentTimeMillis();  

  64.          System.out.println("执行waitForConfirmsOrDie耗费时间: "+(System.currentTimeMillis()-start)+"ms");  

  65.      } catch (Exception e) {  

  66.          e.printStackTrace();  

  67.      }  

  68.  }

  69. }

    在第 50 行调用 Channel 信道的 confirmSelect 方法将当前信道设置成了 confirm 模式，第 57 行通过 for 循环调用 Channel 的 basicPublish 方法发送了 5 条消息到消息队列中，第 58 行调用 waitForConfirms 方法等待 broker 服务端返回 ack 或者 nack 消息，这种模式每发送一条消息就会等待 broker 代理服务器返回消息，这点我们可以从抓包的角度观察结果：

   可以看到上面生产者通过Confirm.Select将当前Channel信道设置成confirm模式，broker代理服务器收到之后回传Confirm.Select-Ok同一将当前Channel设置成confirm模式，此外看到返回5条Basic.Ack消息；

    测试2：批量confirm模式

    这种模式生产者不是每发送一条就等待broker确认，而是发送一批，实现代码见下：

[java] view plain copy

  1. public class ProducerTest {

  2.  public static void main(String[] args) {  

  3.      String exchangeName = "confirmExchange";  

  4.      String queueName = "confirmQueue";  

  5.      String routingKey = "confirmRoutingKey";  

  6.      String bindingKey = "confirmRoutingKey";  

  7.      int count = 100;  

  9.      ConnectionFactory factory = new ConnectionFactory();  

  10.      factory.setHost("172.16.151.74");  

  11.      factory.setUsername("test");  

  12.      factory.setPassword("test");  

  13.      factory.setPort(5672);  

  15.      //创建生产者  

  16.      Sender producer = new Sender(factory, count, exchangeName, queueName,routingKey,bindingKey);  

  17.      producer.run();  

  18.  }

  19. }

  21. class Sender

  22. {

  23.  private ConnectionFactory factory;  

  24.  private int count;  

  25.  private String exchangeName;  

  26.  private String  queueName;  

  27.  private String routingKey;  

  28.  private String bindingKey;  

  30.  public Sender(ConnectionFactory factory,int count,String exchangeName,String queueName,String routingKey,String bindingKey) {  

  31.      this.factory = factory;  

  32.      this.count = count;  

  33.      this.exchangeName = exchangeName;  

  34.      this.queueName = queueName;  

  35.      this.routingKey = routingKey;  

  36.      this.bindingKey = bindingKey;  

  37.  }  

  39.  public void run() {  

  40.      Channel channel = null;  

  41.      try {  

  42.          Connection connection = factory.newConnection();  

  43.          channel = connection.createChannel();  

  44.          //创建exchange  

  45.          channel.exchangeDeclare(exchangeName, "direct", true, false, null);  

  46.          //创建队列  

  47.          channel.queueDeclare(queueName, true, false, false, null);  

  48.          //绑定exchange和queue  

  49.          channel.queueBind(queueName, exchangeName, bindingKey);  

  50.          channel.confirmSelect();  

  51.          //发送持久化消息  

  52.          for(int i = 0;i < count;i++)  

  53.          {  

  54.              //第一个参数是exchangeName(默认情况下代理服务器端是存在一个""名字的exchange的,  

  55.              //因此如果不创建exchange的话我们可以直接将该参数设置成"",如果创建了exchange的话  

  56.              //我们需要将该参数设置成创建的exchange的名字),第二个参数是路由键  

  57.              channel.basicPublish(exchangeName, routingKey,MessageProperties.PERSISTENT_BASIC, ("第"+(i+1)+"条消息").getBytes());  

  58.          }  

  59.          long start = System.currentTimeMillis();  

  60.          channel.waitForConfirmsOrDie();  

  61.          System.out.println("执行waitForConfirmsOrDie耗费时间: "+(System.currentTimeMillis()-start)+"ms");  

  62.      } catch (Exception e) {  

  63.          e.printStackTrace();  

  64.      }  

  65.  }

  66. }

    第 50 行调用 channel.confirmSelect 将当前 channel 信道设置成 confirm 模式，接着在第 57 行通过 for 循环发送了 100 条消息，第 60 行调用了 channel 的 waitForConfirmsOrDie，从 waitForConfirmsOrDie 方法的注释上可以看出，该方法会等到最后一条消息得到确认或者得到 nack 才会结束，也就是说在 waitForConfirmsOrDie 处会造成当前程序的阻塞，以测试 1 程序发送 100 条消息为例，阻塞时间是 135ms，我们再来看看对测试 1 的抓包情况：

   从红色箭头的标号1出可以看到：首先是24向74发送了Confirm.Select消息表示请求将当前信道设置为confirm模式，接着74向24回送了Confirm.Select-Ok消息表示同意将信道设置成confirm模式，从红色标号2处NoWait字段的值为false也印证了我们如果直接调用Channel信道的confirmSelect()方法的话，实际上默认是开启broker回传Confirm.Select-Ok确认消息的；  

   接下来我们看看broker回传给客户端的确认消息数据包是什么样子的呢？同样通过抓包看看结果：


   你会发现，在上面测试1中我们通过for循环发送了100条消息，但是在抓包的时候我们仅仅看到有两个Basic.Ack确认消息回传回来，原因在于上面截图的标号3处，你会发现Multiple域的值是True的，之前我们已经讲过broker可以设置Multiple域表示broker已经收到当前确认消息的Delivery-Tag域之前标号的消息，以上面截图为例的话表示broker告诉发送者编号4之前的消息已经全部收到了，从这点我们看出broker端默认情况下是进行批量回复的，并不是针对每条消息都发送一条ack消息；

   测试2：

   测试1我们仅仅是测试发送者能够收到broker的确认消息以及知道了broker对消息默认是采用批量回复方式的，那么在程序中我们该怎么获取到broker回传回来的确认消息呢，假如我们有时候需要在收到确认消息之后做一些提示性操作该怎么办呢？测试1中，我们采用的是Channel信道的waitForConfirmsOrDie等待broker端回传回ack确认消息的，但我们没法拿到这个ack消息进行后期操作，要想拿到ack消息的话，我们可以给当前Channel信道绑定监听器，具体来说就是调用Channel信道的addConfirmListener方法进行设置，Channel信道在收到broker的ack消息之后会回调设置在该信道监听器上的handleAck方法，在收到nack消息之后会回调设置在该信道监听器上的handleNack方法。

   实现代码：

[java] view plain copy

  1. public class ProducerTest {

  2.  public static void main(String[] args) {  

  3.      String exchangeName = "confirmExchange";  

  4.      String queueName = "confirmQueue";  

  5.      String routingKey = "confirmRoutingKey";  

  6.      String bindingKey = "confirmRoutingKey";  

  7.      int count = 100;  

  9.      ConnectionFactory factory = new ConnectionFactory();  

  10.      factory.setHost("172.16.151.74");  

  11.      factory.setUsername("test");  

  12.      factory.setPassword("test");  

  13.      factory.setPort(5672);  

  15.      //创建生产者  

  16.      Sender producer = new Sender(factory, count, exchangeName, queueName,routingKey,bindingKey);  

  17.      producer.run();  

  18.  }

  19. }

  21. class Sender

  22. {

  23.  private ConnectionFactory factory;  

  24.  private int count;  

  25.  private String exchangeName;  

  26.  private String  queueName;  

  27.  private String routingKey;  

  28.  private String bindingKey;  

  30.  public Sender(ConnectionFactory factory,int count,String exchangeName,String queueName,String routingKey,String bindingKey) {  

  31.      this.factory = factory;  

  32.      this.count = count;  

  33.      this.exchangeName = exchangeName;  

  34.      this.queueName = queueName;  

  35.      this.routingKey = routingKey;  

  36.      this.bindingKey = bindingKey;  

  37.  }  

  39.  public void run() {  

  40.      Channel channel = null;  

  41.      try {  

  42.          Connection connection = factory.newConnection();  

  43.          channel = connection.createChannel();  

  44.          //创建exchange  

  45.          channel.exchangeDeclare(exchangeName, "direct", true, false, null);  

  46.          //创建队列  

  47.          channel.queueDeclare(queueName, true, false, false, null);  

  48.          //绑定exchange和queue  

  49.          channel.queueBind(queueName, exchangeName, bindingKey);  

  50.          channel.confirmSelect();  

  51.          //发送持久化消息  

  52.          for(int i = 0;i < count;i++)  

  53.          {  

  54.              //第一个参数是exchangeName(默认情况下代理服务器端是存在一个""名字的exchange的,  

  55.              //因此如果不创建exchange的话我们可以直接将该参数设置成"",如果创建了exchange的话  

  56.              //我们需要将该参数设置成创建的exchange的名字),第二个参数是路由键  

  57.              channel.basicPublish(exchangeName, routingKey,MessageProperties.PERSISTENT_BASIC, ("第"+(i+1)+"条消息").getBytes());  

  58.          }  

  59.          long start = System.currentTimeMillis();  

  60.          channel.addConfirmListener(new ConfirmListener() {  

  62.              @Override  

  63.              public void handleNack(long deliveryTag, boolean multiple) throws IOException {  

  64.                  System.out.println("nack: deliveryTag = "+deliveryTag+" multiple: "+multiple);  

  65.              }  

  67.              @Override  

  68.              public void handleAck(long deliveryTag, boolean multiple) throws IOException {  

  69.                  System.out.println("ack: deliveryTag = "+deliveryTag+" multiple: "+multiple);  

  70.              }  

  71.          });  

  72.          System.out.println("执行waitForConfirmsOrDie耗费时间: "+(System.currentTimeMillis()-start)+"ms");  

  73.      } catch (Exception e) {  

  74.          e.printStackTrace();  

  75.      }  

  76.  }

  77. }

    第 60 行我们调用了 Channel 信道的 addConfirmListener 设置了监听器，并且在监听器的 handleAck 和 handleNack 方法中打印了信息，运行程序查看输出：

   可以看到，虽然我们还是发送了100条消息，同样我们并没有收到100个ack消息 ，只收到两个ack消息，并且这两个ack消息的multiple域都为true，这点和测试1是相同的，你多次运行程序会发现每次发送回来的ack消息中的deliveryTag域的值并不是一样的，说明broker端批量回传给发送者的ack消息并不是以固定的批量大小回传的；

   也就是我们通过信道Channel的waitForConfirmsOrDie方法或者为信道设置监听器都可以保证发送者收到broker回传的ack或者nack消息，那么这两种方式有什么区别呢？从测试一的第61行代码以及测试2的第72行代码处你就能找到答案啦，测试1中调用waitForConfirmsOrDie方法发送100条消息并且全部收到确认需要135ms，测试2中通过监听器的方式仅仅需要1ms，说明调用waitForConfirmsOrDie会造成程序的阻塞，通过监听器并不会造成程序的阻塞，下一篇博客我会试着从RabbitMQ的源码层面来分析这两种方式造成这种区别的原因啦啦；

   参考资料：

   [RabbitMQ官网](http://www.rabbitmq.com/confirms.html)

   [RabbitMQ不同Confirm模式下的性能对比](http://ju.outofmemory.cn/entry/177937)
  • RabbitMQ

    RabbitMQ 是一个开源的 AMQP 实现，服务器端用 Erlang 语言编写，支持多种语言客户端，如：Python、Ruby、.NET、Java、C、PHP、ActionScript 等。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

    49 引用 • 60 回帖 • 399 关注
574 2 105 115 17 42 6 49 229

相关帖子

回帖
深入学习 RabbitMQ(三)：channel 的 confirm 模式

欢迎来到这里！

我们正在构建一个小众社区，大家在这里相互信任，以平等 • 自由 • 奔放的价值观进行分享交流。最终，希望大家能够找到与自己志同道合的伙伴，共同成长。

注册 关于
请输入回帖内容 ...
honglan
回帖
0
 帖子
28
 积分
621

近期热议

推荐标签 标签

  • 以太坊

    以太坊（Ethereum）并不是一个机构，而是一款能够在区块链上实现智能合约、开源的底层系统。以太坊是一个平台和一种编程语言 Solidity，使开发人员能够建立和发布下一代去中心化应用。 以太坊可以用来编程、分散、担保和交易任何事物：投票、域名、金融交易所、众筹、公司管理、合同和知识产权等等。

    34 引用 • 367 回帖 • 1 关注
  • abitmean

    有点意思就行了

    • 23 关注
  • 微软

    微软是一家美国跨国科技公司，也是世界 PC 软件开发的先导，由比尔·盖茨与保罗·艾伦创办于 1975 年，公司总部设立在华盛顿州的雷德蒙德（Redmond，邻近西雅图）。以研发、制造、授权和提供广泛的电脑软件服务业务为主。

    8 引用 • 44 回帖
  • C++

    C++ 是在 C 语言的基础上开发的一种通用编程语言，应用广泛。C++ 支持多种编程范式，面向对象编程、泛型编程和过程化编程。

    106 引用 • 152 回帖
  • 大数据

    大数据（big data）是指无法在一定时间范围内用常规软件工具进行捕捉、管理和处理的数据集合，是需要新处理模式才能具有更强的决策力、洞察发现力和流程优化能力的海量、高增长率和多样化的信息资产。

    89 引用 • 113 回帖
  • Vditor

    Vditor 是一款浏览器端的 Markdown 编辑器，支持所见即所得、即时渲染（类似 Typora）和分屏预览模式。它使用 TypeScript 实现，支持原生 JavaScript、Vue、React 和 Angular。

    313 引用 • 1667 回帖 • 1 关注
  • Flume

    Flume 是一套分布式的、可靠的，可用于有效地收集、聚合和搬运大量日志数据的服务架构。

    9 引用 • 6 回帖 • 594 关注
  • 服务器

    服务器，也称伺服器，是提供计算服务的设备。由于服务器需要响应服务请求，并进行处理，因此一般来说服务器应具备承担服务并且保障服务的能力。

    124 引用 • 580 回帖
  • 域名

    域名（Domain Name），简称域名、网域，是由一串用点分隔的名字组成的 Internet 上某一台计算机或计算机组的名称，用于在数据传输时标识计算机的电子方位（有时也指地理位置）。

    43 引用 • 208 回帖 • 1 关注
  • 博客

    记录并分享人生的经历。

    270 引用 • 2386 回帖
  • 房星科技

    房星网，我们不和没有钱的程序员谈理想，我们要让程序员又有理想又有钱。我们有雄厚的房地产行业线下资源，遍布昆明全城的 100 家门店、四千地产经纪人是我们坚实的后盾。

    6 引用 • 141 回帖 • 559 关注
  • 倾城之链
    23 引用 • 66 回帖 • 102 关注
  • Unity

    Unity 是由 Unity Technologies 开发的一个让开发者可以轻松创建诸如 2D、3D 多平台的综合型游戏开发工具，是一个全面整合的专业游戏引擎。

    25 引用 • 7 回帖 • 245 关注
  • SMTP

    SMTP（Simple Mail Transfer Protocol）即简单邮件传输协议，它是一组用于由源地址到目的地址传送邮件的规则，由它来控制信件的中转方式。SMTP 协议属于 TCP/IP 协议簇，它帮助每台计算机在发送或中转信件时找到下一个目的地。

    4 引用 • 18 回帖 • 589 关注
  • Webswing

    Webswing 是一个能将任何 Swing 应用通过纯 HTML5 运行在浏览器中的 Web 服务器，详细介绍请看 将 Java Swing 应用变成 Web 应用 。

    1 引用 • 15 回帖 • 635 关注
  • wolai

    我来 wolai：不仅仅是未来的云端笔记！

    1 引用 • 11 回帖 • 2 关注
  • WebComponents

    Web Components 是 W3C 定义的标准，它给了前端开发者扩展浏览器标签的能力，可以方便地定制可复用组件，更好的进行模块化开发，解放了前端开发者的生产力。

    1 引用 • 25 关注
  • HBase

    HBase 是一个分布式的、面向列的开源数据库，该技术来源于 Fay Chang 所撰写的 Google 论文 “Bigtable：一个结构化数据的分布式存储系统”。就像 Bigtable 利用了 Google 文件系统所提供的分布式数据存储一样，HBase 在 Hadoop 之上提供了类似于 Bigtable 的能力。

    17 引用 • 6 回帖 • 45 关注
  • iOS

    iOS 是由苹果公司开发的移动操作系统，最早于 2007 年 1 月 9 日的 Macworld 大会上公布这个系统，最初是设计给 iPhone 使用的，后来陆续套用到 iPod touch、iPad 以及 Apple TV 等产品上。iOS 与苹果的 Mac OS X 操作系统一样，属于类 Unix 的商业操作系统。

    84 引用 • 139 回帖
  • 30Seconds

    📙 前端知识精选集，包含 HTML、CSS、JavaScript、React、Node、安全等方面，每天仅需 30 秒。

    • 精选常见面试题，帮助您准备下一次面试
    • 精选常见交互，帮助您拥有简洁酷炫的站点
    • 精选有用的 React 片段，帮助你获取最佳实践
    • 精选常见代码集，帮助您提高打码效率
    • 整理前端界的最新资讯，邀您一同探索新世界
    488 引用 • 383 回帖 • 5 关注
  • Lute

    Lute 是一款结构化的 Markdown 引擎，支持 Go 和 JavaScript。

    25 引用 • 191 回帖 • 21 关注
  • Dubbo

    Dubbo 是一个分布式服务框架，致力于提供高性能和透明化的 RPC 远程服务调用方案，是 [阿里巴巴] SOA 服务化治理方案的核心框架，每天为 2,000+ 个服务提供 3,000,000,000+ 次访问量支持，并被广泛应用于阿里巴巴集团的各成员站点。

    60 引用 • 82 回帖 • 609 关注
  • 负能量

    上帝为你关上了一扇门，然后就去睡觉了....努力不一定能成功，但不努力一定很轻松 (° ー °〃)

    85 引用 • 1201 回帖 • 449 关注
  • 区块链

    区块链是分布式数据存储、点对点传输、共识机制、加密算法等计算机技术的新型应用模式。所谓共识机制是区块链系统中实现不同节点之间建立信任、获取权益的数学算法 。

    91 引用 • 751 回帖
  • Bootstrap

    Bootstrap 是 Twitter 推出的一个用于前端开发的开源工具包。它由 Twitter 的设计师 Mark Otto 和 Jacob Thornton 合作开发，是一个 CSS / HTML 框架。

    18 引用 • 33 回帖 • 684 关注
  • Ubuntu

    Ubuntu（友帮拓、优般图、乌班图）是一个以桌面应用为主的 Linux 操作系统，其名称来自非洲南部祖鲁语或豪萨语的“ubuntu”一词，意思是“人性”、“我的存在是因为大家的存在”，是非洲传统的一种价值观，类似华人社会的“仁爱”思想。Ubuntu 的目标在于为一般用户提供一个最新的、同时又相当稳定的主要由自由软件构建而成的操作系统。

    123 引用 • 168 回帖
  • Facebook

    Facebook 是一个联系朋友的社交工具。大家可以通过它和朋友、同事、同学以及周围的人保持互动交流，分享无限上传的图片，发布链接和视频，更可以增进对朋友的了解。

    4 引用 • 15 回帖 • 455 关注

最新标签