2013年10月14日 (一) 10:51的版本

服务简介

微博平台订阅服务提高了应用获取微博数据的效率。根据应用指定的订阅条件，如：关键词、用户、域名等，平台主动将新产生的数据实时推送给应用，应用不需要轮询请求接口。

订阅服务优点如下：

1、将新数据实时推送给应用；

2、节省网络开销；

3、提供数据量更多更全；

4、提高应用访问接口的效率；

订阅条件

订阅用户：

开发者可最多指定10000个用户。当订阅微博时，微博为指定用户发出的，则推送；当订阅评论时，评论对应的微博为指定用户产生的评论则推送；若不指定，则无此限制。

订阅关键词：

开发者可最多指定400个关键词。当订阅微博时，微博中含有指定关键词的，则推送；当订阅评论时，评论对应的微博含有此关键词的评论则推送；若不指定，则无此限制。

订阅域名：

开发者可最多指定20个域名。当订阅微博时，微博中包含短链所对应原始链接为指定域名下的，则推送；当订阅评论时，评论对应的微博内包含短链所对应原始链接为指定域名的评论则推送；若不指定，则无此限制。

订阅应用：

开发者可指定只推送生成订阅的应用产生的数据，若不指定则推送所有应用的数据。

订阅媒体类型：

当订阅微博时，才可以指定该条件。开发者可指定原创、转发、视频、音乐、或图片类型。则推送指定类型的微博；若不指定，则无此限制；当订阅评论时，无此筛选条件。

订阅数据类型：

开发者可指定推送微博、或评论数据；若不指定，则默认推送微博。

订阅百分比：

开发者可指定满足以上订阅条件的数据的百分比，若不指定，则推送满足条件的1%数据。

订阅时间范围：

开发者可指定推送开始时间和结束时间。在指定开始时间推送服务准备就绪，结束时间终止推送。若不指定，则开始时间默认等于订阅生成时间，一直推送。在推送服务准备就绪时，开发者可以调用接口进行连接，从而接收数据。

使用步骤

付费订阅服务使用步骤如下：

• 1、生成订阅：
  • 开发者线下填写附的订阅服务申请单，填写订阅条件等信息，平台人员根据申请单信息，生成订阅。每个应用可以有多个订阅。
• 2、推送服务就绪：
  • 开发者若在申请单中指定了推送开始时间，则在指定开始时间该订阅的推送服务就绪；若没有指定，则订阅生成后推送服务就已经就绪。
• 3、应用连接、推送开始：
  • 若订阅已经生成，但推送服务未就绪，则开发者订阅管理后台的推送状态显示：准备中；若推送服务已经就绪，则显示：准备就绪。在准备就绪状态下，应用才可以调用接口进行连接，接收数据。否则，调用接口时报错。调用接口如下：
    • a.、订阅微博，调用如下接口：
      • 接口名称：datapush/status
      • 调用该接口进行连接，接收微博数据。
      • URL：http://180.149.153.38/datapush/status or http://180.149.153.39/datapush/status
      • HTTP Request Method：
      • GET
      • Requires Authentication：
      • False

Parameters：

• • • b、订阅评论，调用如下接口：
      • 接口名称：datapush/comment
      • 调用该接口进行连接，接收评论数据。
      • URL：http://180.149.153.40/datapush/comment or http://180.149.153.41/datapush/comment
      • HTTP Request Method：
      • GET
      • Requires Authentication：
      • False

Parameters：

接口调用java代码示例如下：

//从datapush服务器获取连接

         //streaming server列表，可以配置，当前sever 180.149.153.38，180.149.153.39微博

         private List<String> streamingUrlList;

         //当前连接的streaming server，可以配置

         private int curStreamUrlIndex =0;

         private GetMethod connectStreamServer() throws StreamingException {

                   String targetURL = streamingUrlList.get(curStreamUrlIndex);

                   //从指定的location开始读取数据，保证读取数据的连续性，消息完整性

                   //链接断开的重新链接机制

                   if(lastMsgLocation > 0){

                            targetURL += "&loc=" + lastMsgLocation;

                   }

                   logger.info("StreamingReceiver http get url=" + targetURL);

                   GetMethod method = new GetMethod(targetURL);

                   int statusCode;

                   try {

                            statusCode = httpClient.executeMethod(method);

                   }

                   catch(Exception e){

                            //dealException;

                   }

                   if (statusCode != HttpStatus.SC_OK) {

                            throw new StreamingException(".. ");

                   }                          

                   try {

                            inputStream = new                   DataInputStream(method.getResponseBodyAsStream());

                   } catch (IOException e) {

                            throw new StreamingException("get stream input io exception", e);

                   }                

                   return method;

         }

//开一个线程从服务器读取数据：

public void run() {

                   while(true){

                            GetMethod method=null;

                            recIndex = 0;

                            recBuf = new byte[recBufSize];

                            try{

                                     method = connectStreamServer();

                                     while(true){

                                               processLine();

                                     }

                            }catch(Exception e){

                   //当连接断开时，重新连接

                   logger.error("streaming process error ",e);

                   curStreamUrlIndex=++curStreamUrlIndex % streamingUrlList.size();

                            }finally{

                                               method.releaseConnection();                  

                                     }

                            }

                   }

         }

         private void processLine() throws IOException{

                   byte[] bytes = readLineBytes();

                   if(bytes != null && bytes.length > 0){

                            String message = new String(bytes);

                                     //handle the message you get

                                     handleMessage() ；

                                     lastMsgLocation=Message.getId();

                            }

                   }

         }

         public byte[] readLineBytes() throws IOException {

                   byte[] result = null;

                   ByteArrayOutputStream bos = new ByteArrayOutputStream();

                   int readCount = 0;

                   if (recIndex > 0 && read(bos)) {

                            return bos.toByteArray();

                   }

                   while ((readCount = inputStream.read(recBuf, recIndex, recBuf.length

                                     - recIndex)) > 0) {

                            recIndex = recIndex + readCount;

                            if (read(bos)) {

                                     break;

                            }

                   }

                   result = bos.toByteArray();

                   if (result == null

                                     || (result != null && result.length <= 0 && recIndex <= 0)) {

                            throw new IOException(

                            "++++ Stream appears to be dead, so closing it down");

                   }

                   return result;

         }

         private boolean read(ByteArrayOutputStream bos) {

                   boolean result = false;

                   int index = -1;

                   for (int i = 0; i < recIndex - 1; i++) {

                            if (recBuf[i] == 13 && recBuf[i + 1] == 10) {//13 cr-回车 10 lf-换行

                                     index = i;

                                     break;

                            }

                   }

                   if (index >= 0) {

                            bos.write(recBuf, 0, index);

                            byte[] newBuf = new byte[recBufSize];

                            if (recIndex > index + 2) {

                                     System.arraycopy(recBuf, index + 2, newBuf, 0, recIndex

                                                        - index - 2);

                            }

                            recBuf = newBuf;

                            recIndex = recIndex - index - 2;

                            result = true;

                   } else {

                            if (recBuf[recIndex - 1] == 13) {

                                     bos.write(recBuf, 0, recIndex - 1);

                                     Arrays.fill(recBuf,(byte)0);

                                     recBuf[0] = 13;

                                     recIndex = 1;

                            } else {

                                     bos.write(recBuf, 0, recIndex);

                                     Arrays.fill(recBuf,(byte)0);

                                     recIndex = 0;

                            }

  &n