AWS LAMBDA：Java的：室壁運動活動

Question

我有麻煩找的什麼，我試圖做例子......

我想創建Java中的lambda函數。我認爲我總是使用Javascript來實現Lambda函數，但在這種情況下，我最終會重新使用已經用Java編寫的應用程序邏輯，所以這很有意義。

在過去，我已經編寫了由Kinesis事件觸發的JavaScript Lambda函數。超級簡單，函數接收事件作爲參數，做一些事情，瞧。我想用Java來做同樣的事情。很簡單：

室壁運動事件（S） - >觸發功能 - >（Java）的接收室壁運動活動，與他們做一些

任何人有這樣的使用案例的經驗嗎？

Answer 1

下面是我寫的一些示例代碼，用於在內部展示相同的概念。此代碼將事件從一個流轉發到另一個流。

注意如果在轉發這個錯誤代碼不處理重試，也不是意味着要在生產環境中高性能，但它確實演示瞭如何從發佈流處理的記錄。

import com.amazonaws.regions.Region; 
import com.amazonaws.regions.Regions; 
import com.amazonaws.services.kinesis.AmazonKinesisClient; 
import com.amazonaws.services.kinesis.model.PutRecordsRequest; 
import com.amazonaws.services.kinesis.model.PutRecordsRequestEntry; 
import com.amazonaws.services.kinesis.model.PutRecordsResult; 
import com.amazonaws.services.lambda.runtime.Context; 
import com.amazonaws.services.lambda.runtime.LambdaLogger; 
import com.amazonaws.services.lambda.runtime.events.KinesisEvent; 

import java.nio.ByteBuffer; 
import java.util.ArrayList; 
import java.util.List; 
import java.util.Map; 
import java.util.stream.Collectors; 

public class KinesisToKinesis { 

    private LambdaLogger logger; 
    final private AmazonKinesisClient kinesisClient = new AmazonKinesisClient(); 

    public PutRecordsResult eventHandler(KinesisEvent event, Context context) { 
     logger = context.getLogger(); 
     if (event == null || event.getRecords() == null) { 
      logger.log("Event contains no data" + System.lineSeparator()); 
      return null; 
     } else { 
      logger.log("Received " + event.getRecords().size() + 
       " records from " + event.getRecords().get(0).getEventSourceARN() + System.lineSeparator()); 
     } 

     final Long startTime = System.currentTimeMillis(); 

     // set up the client 
     Region region; 
     final Map<String, String> environmentVariables = System.getenv(); 
     if (environmentVariables.containsKey("AWS_REGION")) { 
      region = Region.getRegion(Regions.fromName(environmentVariables.get("AWS_REGION"))); 
     } else { 
      region = Region.getRegion(Regions.US_WEST_2); 
      logger.log("Using default region: " + region.toString() + System.lineSeparator()); 
     } 
     kinesisClient.setRegion(region); 

     Long elapsed = System.currentTimeMillis() - startTime; 
     logger.log("Finished setup in " + elapsed + " ms" + System.lineSeparator()); 

     PutRecordsRequest putRecordsRequest = new PutRecordsRequest().withStreamName("usagecounters-global"); 
     List<PutRecordsRequestEntry> putRecordsRequestEntryList = event.getRecords().parallelStream() 
      .map(r -> new PutRecordsRequestEntry() 
        .withData(ByteBuffer.wrap(r.getKinesis().getData().array())) 
        .withPartitionKey(r.getKinesis().getPartitionKey())) 
      .collect(Collectors.toList()); 

     putRecordsRequest.setRecords(putRecordsRequestEntryList); 

     elapsed = System.currentTimeMillis() - startTime; 
     logger.log("Processed " + putRecordsRequest.getRecords().size() + 
      " records in " + elapsed + " ms" + System.lineSeparator()); 

     PutRecordsResult putRecordsResult = kinesisClient.putRecords(putRecordsRequest); 
     elapsed = System.currentTimeMillis() - startTime; 
     logger.log("Forwarded " + putRecordsRequest.getRecords().size() + 
       " records to Kinesis " + putRecordsRequest.getStreamName() + 
       " in " + elapsed + " ms" + System.lineSeparator()); 
     return putRecordsResult; 
    } 
}